fdomain_client/
lib.rs

1// Copyright 2024 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 fidl_fuchsia_fdomain as proto;
6use fidl_message::TransactionHeader;
7use fuchsia_async as _;
8use fuchsia_sync::Mutex;
9use futures::FutureExt;
10use futures::channel::oneshot::Sender as OneshotSender;
11use futures::stream::Stream as StreamTrait;
12use std::collections::{HashMap, VecDeque};
13use std::convert::Infallible;
14use std::future::Future;
15use std::num::NonZeroU32;
16use std::pin::Pin;
17use std::sync::{Arc, LazyLock};
18use std::task::{Context, Poll, Waker, ready};
19
20mod channel;
21mod event;
22mod event_pair;
23mod handle;
24mod responder;
25mod socket;
26
27#[cfg(test)]
28mod test;
29
30pub mod fidl;
31pub mod fidl_next;
32
33use responder::Responder;
34
35pub use channel::{
36    AnyHandle, Channel, ChannelMessageStream, ChannelWriter, HandleInfo, HandleOp, MessageBuf,
37};
38pub use event::Event;
39pub use event_pair::Eventpair as EventPair;
40pub use handle::unowned::Unowned;
41pub use handle::{
42    AsHandleRef, Handle, HandleBased, HandleRef, NullableHandle, OnFDomainSignals, Peered,
43};
44pub use proto::{Error as FDomainError, WriteChannelError, WriteSocketError};
45pub use socket::{Socket, SocketDisposition, SocketReadStream, SocketWriter};
46
47// Unsupported handle types.
48#[rustfmt::skip]
49pub use Handle as Clock;
50#[rustfmt::skip]
51pub use Handle as Exception;
52#[rustfmt::skip]
53pub use Handle as Fifo;
54#[rustfmt::skip]
55pub use Handle as Iob;
56#[rustfmt::skip]
57pub use Handle as Job;
58#[rustfmt::skip]
59pub use Handle as Process;
60#[rustfmt::skip]
61pub use Handle as Resource;
62#[rustfmt::skip]
63pub use Handle as Stream;
64#[rustfmt::skip]
65pub use Handle as Thread;
66#[rustfmt::skip]
67pub use Handle as Vmar;
68#[rustfmt::skip]
69pub use Handle as Vmo;
70#[rustfmt::skip]
71pub use Handle as Counter;
72
73use proto::f_domain_ordinals as ordinals;
74
75fn write_fdomain_error(error: &FDomainError, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76    match error {
77        FDomainError::TargetError(e) => {
78            let e = zx_status::Status::from_raw(*e);
79            write!(f, "Target-side error {e}")
80        }
81        FDomainError::BadHandleId(proto::BadHandleId { id }) => {
82            write!(f, "Tried to use invalid handle id {id}")
83        }
84        FDomainError::WrongHandleType(proto::WrongHandleType { expected, got }) => write!(
85            f,
86            "Tried to use handle as {expected:?} but target reported handle was of type {got:?}"
87        ),
88        FDomainError::StreamingReadInProgress(proto::StreamingReadInProgress {}) => {
89            write!(f, "Handle is occupied delivering streaming reads")
90        }
91        FDomainError::NoReadInProgress(proto::NoReadInProgress {}) => {
92            write!(f, "No streaming read was in progress")
93        }
94        FDomainError::NewHandleIdOutOfRange(proto::NewHandleIdOutOfRange { id }) => {
95            write!(
96                f,
97                "Tried to create a handle with id {id}, which is outside the valid range for client handles"
98            )
99        }
100        FDomainError::NewHandleIdReused(proto::NewHandleIdReused { id, same_call }) => {
101            if *same_call {
102                write!(f, "Tried to create two or more new handles with the same id {id}")
103            } else {
104                write!(
105                    f,
106                    "Tried to create a new handle with id {id}, which is already the id of an existing handle"
107                )
108            }
109        }
110        FDomainError::WroteToSelf(proto::WroteToSelf {}) => {
111            write!(f, "Tried to write a channel into itself")
112        }
113        FDomainError::ClosedDuringRead(proto::ClosedDuringRead {}) => {
114            write!(f, "Handle closed while being read")
115        }
116        _ => todo!(),
117    }
118}
119
120/// Result type alias.
121pub type Result<T, E = Error> = std::result::Result<T, E>;
122
123/// Error type emitted by FDomain operations.
124#[derive(Clone)]
125pub enum Error {
126    SocketWrite(WriteSocketError),
127    ChannelWrite(WriteChannelError),
128    FDomain(FDomainError),
129    Protocol(::fidl::Error),
130    ProtocolObjectTypeIncompatible,
131    ProtocolRightsIncompatible,
132    ProtocolSignalsIncompatible,
133    ProtocolStreamEventIncompatible,
134    Transport(Option<Arc<std::io::Error>>),
135    ConnectionMismatch,
136    StreamingAborted,
137}
138
139impl std::fmt::Display for Error {
140    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
141        match self {
142            Self::SocketWrite(proto::WriteSocketError { error, wrote }) => {
143                write!(f, "While writing socket (after {wrote} bytes written successfully): ")?;
144                write_fdomain_error(error, f)
145            }
146            Self::ChannelWrite(proto::WriteChannelError::Error(error)) => {
147                write!(f, "While writing channel: ")?;
148                write_fdomain_error(error, f)
149            }
150            Self::ChannelWrite(proto::WriteChannelError::OpErrors(errors)) => {
151                write!(f, "Couldn't write all handles into a channel:")?;
152                for (pos, error) in
153                    errors.iter().enumerate().filter_map(|(num, x)| x.as_ref().map(|y| (num, &**y)))
154                {
155                    write!(f, "\n  Handle in position {pos}: ")?;
156                    write_fdomain_error(error, f)?;
157                }
158                Ok(())
159            }
160            Self::ProtocolObjectTypeIncompatible => {
161                write!(f, "The FDomain protocol does not recognize an object type")
162            }
163            Self::ProtocolRightsIncompatible => {
164                write!(f, "The FDomain protocol does not recognize some rights")
165            }
166            Self::ProtocolSignalsIncompatible => {
167                write!(f, "The FDomain protocol does not recognize some signals")
168            }
169            Self::ProtocolStreamEventIncompatible => {
170                write!(f, "The FDomain protocol does not recognize a received streaming IO event")
171            }
172            Self::FDomain(e) => write_fdomain_error(e, f),
173            Self::Protocol(e) => write!(f, "Protocol error: {e}"),
174            Self::Transport(Some(e)) => write!(f, "Transport error: {e:?}"),
175            Self::Transport(None) => write!(f, "Transport closed"),
176            Self::ConnectionMismatch => {
177                write!(f, "Tried to use an FDomain handle from a different connection")
178            }
179            Self::StreamingAborted => write!(f, "This channel is no longer streaming"),
180        }
181    }
182}
183
184impl std::fmt::Debug for Error {
185    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
186        match self {
187            Self::SocketWrite(e) => f.debug_tuple("SocketWrite").field(e).finish(),
188            Self::ChannelWrite(e) => f.debug_tuple("ChannelWrite").field(e).finish(),
189            Self::FDomain(e) => f.debug_tuple("FDomain").field(e).finish(),
190            Self::Protocol(e) => f.debug_tuple("Protocol").field(e).finish(),
191            Self::Transport(e) => f.debug_tuple("Transport").field(e).finish(),
192            Self::ProtocolObjectTypeIncompatible => write!(f, "ProtocolObjectTypeIncompatible "),
193            Self::ProtocolRightsIncompatible => write!(f, "ProtocolRightsIncompatible "),
194            Self::ProtocolSignalsIncompatible => write!(f, "ProtocolSignalsIncompatible "),
195            Self::ProtocolStreamEventIncompatible => write!(f, "ProtocolStreamEventIncompatible"),
196            Self::ConnectionMismatch => write!(f, "ConnectionMismatch"),
197            Self::StreamingAborted => write!(f, "StreamingAborted"),
198        }
199    }
200}
201
202impl std::error::Error for Error {}
203
204impl From<FDomainError> for Error {
205    fn from(other: FDomainError) -> Self {
206        Self::FDomain(other)
207    }
208}
209
210impl From<::fidl::Error> for Error {
211    fn from(other: ::fidl::Error) -> Self {
212        Self::Protocol(other)
213    }
214}
215
216impl From<WriteSocketError> for Error {
217    fn from(other: WriteSocketError) -> Self {
218        Self::SocketWrite(other)
219    }
220}
221
222impl From<WriteChannelError> for Error {
223    fn from(other: WriteChannelError) -> Self {
224        Self::ChannelWrite(other)
225    }
226}
227
228/// An error emitted internally by the client. Similar to [`Error`] but does not
229/// contain several variants which are irrelevant in the contexts where it is
230/// used.
231#[derive(Clone)]
232enum InnerError {
233    Protocol(::fidl::Error),
234    ProtocolStreamEventIncompatible,
235    Transport(Option<Arc<std::io::Error>>),
236}
237
238impl From<InnerError> for Error {
239    fn from(other: InnerError) -> Self {
240        match other {
241            InnerError::Protocol(p) => Error::Protocol(p),
242            InnerError::ProtocolStreamEventIncompatible => Error::ProtocolStreamEventIncompatible,
243            InnerError::Transport(t) => Error::Transport(t),
244        }
245    }
246}
247
248impl From<::fidl::Error> for InnerError {
249    fn from(other: ::fidl::Error) -> Self {
250        InnerError::Protocol(other)
251    }
252}
253
254// TODO(399717689) Figure out if we could just use AsyncRead/Write instead of a special trait.
255/// Implemented by objects which provide a transport over which we can speak the
256/// FDomain protocol.
257///
258/// The implementer must provide two things:
259/// 1) An incoming stream of messages presented as `Vec<u8>`. This is provided
260///    via the `Stream` trait, which this trait requires.
261/// 2) A way to send messages. This is provided by implementing the
262///    `poll_send_message` method.
263pub trait FDomainTransport: StreamTrait<Item = Result<Box<[u8]>, std::io::Error>> + Send {
264    /// Attempt to send a message asynchronously. Messages should be sent so
265    /// that they arrive at the target in order.
266    fn poll_send_message(
267        self: Pin<&mut Self>,
268        msg: &[u8],
269        ctx: &mut Context<'_>,
270    ) -> Poll<Result<(), Option<std::io::Error>>>;
271
272    /// Optional debug information outlet.
273    fn debug_fmt(&self, _: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274        Ok(())
275    }
276
277    /// Whether `debug_fmt` does anything.
278    fn has_debug_fmt(&self) -> bool {
279        false
280    }
281}
282
283/// Wrapper for an `FDomainTransport` implementer that:
284/// 1) Provides a queue for outgoing messages so we need not have an await point
285///    when we submit a message.
286/// 2) Drops the transport on error, then returns the last observed error for
287///    all future operations.
288enum Transport {
289    Transport(Pin<Box<dyn FDomainTransport>>, VecDeque<Box<[u8]>>, Vec<Waker>),
290    Error(InnerError),
291}
292
293impl Transport {
294    /// Get the failure mode of the transport if it has failed.
295    fn error(&self) -> Option<InnerError> {
296        match self {
297            Transport::Transport(_, _, _) => None,
298            Transport::Error(inner_error) => Some(inner_error.clone()),
299        }
300    }
301
302    /// Enqueue a message to be sent on this transport.
303    fn push_msg(&mut self, msg: Box<[u8]>) {
304        if let Transport::Transport(_, v, w) = self {
305            v.push_back(msg);
306            w.drain(..).for_each(Waker::wake);
307        }
308    }
309
310    /// Push messages in the send queue out through the transport.
311    fn poll_send_messages(&mut self, ctx: &mut Context<'_>) -> Poll<InnerError> {
312        match self {
313            Transport::Error(e) => Poll::Ready(e.clone()),
314            Transport::Transport(t, v, w) => {
315                while let Some(msg) = v.front() {
316                    match t.as_mut().poll_send_message(msg, ctx) {
317                        Poll::Ready(Ok(())) => {
318                            v.pop_front();
319                        }
320                        Poll::Ready(Err(e)) => {
321                            let e = e.map(Arc::new);
322                            *self = Transport::Error(InnerError::Transport(e.clone()));
323                            return Poll::Ready(InnerError::Transport(e));
324                        }
325                        Poll::Pending => return Poll::Pending,
326                    }
327                }
328
329                if v.is_empty() {
330                    w.push(ctx.waker().clone());
331                } else {
332                    ctx.waker().wake_by_ref();
333                }
334                Poll::Pending
335            }
336        }
337    }
338
339    /// Get the next incoming message from the transport.
340    fn poll_next(&mut self, ctx: &mut Context<'_>) -> Poll<Result<Box<[u8]>, InnerError>> {
341        match self {
342            Transport::Error(e) => Poll::Ready(Err(e.clone())),
343            Transport::Transport(t, _, _) => match ready!(t.as_mut().poll_next(ctx)) {
344                Some(Ok(x)) => Poll::Ready(Ok(x)),
345                Some(Err(e)) => {
346                    let e = Arc::new(e);
347                    *self = Transport::Error(InnerError::Transport(Some(Arc::clone(&e))));
348                    Poll::Ready(Err(InnerError::Transport(Some(e))))
349                }
350                Option::None => Poll::Ready(Err(InnerError::Transport(None))),
351            },
352        }
353    }
354}
355
356impl Drop for Transport {
357    fn drop(&mut self) {
358        if let Transport::Transport(_, _, wakers) = self {
359            wakers.drain(..).for_each(Waker::wake);
360        }
361    }
362}
363
364/// State of a socket that is or has been read from.
365struct SocketReadState {
366    wakers: Vec<Waker>,
367    queued: VecDeque<Result<proto::SocketData, Error>>,
368    read_request_pending: bool,
369    is_streaming: bool,
370}
371
372impl SocketReadState {
373    /// Handle an incoming message, which is either a channel streaming event or
374    /// response to a `ChannelRead` request.
375    fn handle_incoming_message(&mut self, msg: Result<proto::SocketData, Error>) -> Vec<Waker> {
376        self.queued.push_back(msg);
377        std::mem::replace(&mut self.wakers, Vec::new())
378    }
379}
380
381/// State of a channel that is or has been read from.
382struct ChannelReadState {
383    wakers: Vec<Waker>,
384    queued: VecDeque<Result<proto::ChannelMessage, Error>>,
385    read_request_pending: bool,
386    is_streaming: bool,
387}
388
389impl ChannelReadState {
390    /// Handle an incoming message, which is either a channel streaming event or
391    /// response to a `ChannelRead` request.
392    fn handle_incoming_message(&mut self, msg: Result<proto::ChannelMessage, Error>) -> Vec<Waker> {
393        self.queued.push_back(msg);
394        std::mem::replace(&mut self.wakers, Vec::new())
395    }
396}
397
398/// Lock-protected interior of `Client`
399struct ClientInner {
400    transport: Transport,
401    transactions: HashMap<NonZeroU32, responder::Responder>,
402    channel_read_states: HashMap<proto::HandleId, ChannelReadState>,
403    socket_read_states: HashMap<proto::HandleId, SocketReadState>,
404    next_tx_id: u32,
405    waiting_to_close: Vec<proto::HandleId>,
406    waiting_to_close_waker: Waker,
407
408    /// There is a lock around `ClientInner`, and sometimes the FIDL bindings
409    /// give us wakers that want to do handle operations synchronously on wake,
410    /// which means we can double-take the lock if we wake a waker while we hold
411    /// it. This is a place to store wakers that we'd like to be woken as soon
412    /// as we're not holding that lock, to avoid these weird reentrancy issues.
413    wakers_to_wake: Vec<Waker>,
414}
415
416impl ClientInner {
417    /// Serialize and enqueue a new transaction, including header and transaction ID.
418    fn request<S: fidl_message::Body>(&mut self, ordinal: u64, request: S, responder: Responder) {
419        if ordinal != ordinals::CLOSE {
420            self.process_waiting_to_close();
421        }
422        let tx_id = self.next_tx_id;
423
424        let header = TransactionHeader::new(tx_id, ordinal, fidl_message::DynamicFlags::FLEXIBLE);
425        let msg = fidl_message::encode_message(header, request).expect("Could not encode request!");
426        self.next_tx_id += 1;
427        assert!(
428            self.transactions.insert(tx_id.try_into().unwrap(), responder).is_none(),
429            "Allocated same tx id twice!"
430        );
431        self.transport.push_msg(msg.into());
432    }
433
434    fn process_waiting_to_close(&mut self) {
435        if !self.waiting_to_close.is_empty() {
436            let handles = std::mem::replace(&mut self.waiting_to_close, Vec::new());
437            // We've dropped the handle object. Nobody is going to wait to read
438            // the buffers anymore. This is a safe time to drop the read state.
439            for handle in &handles {
440                let _ = self.channel_read_states.remove(handle);
441                let _ = self.socket_read_states.remove(handle);
442            }
443            self.request(
444                ordinals::CLOSE,
445                proto::FDomainCloseRequest { handles },
446                Responder::Ignore,
447            );
448        }
449    }
450
451    /// Polls the underlying transport to ensure any incoming or outgoing
452    /// messages are processed as far as possible. Errors if the transport has failed.
453    fn try_poll_transport(
454        &mut self,
455        ctx: &mut Context<'_>,
456    ) -> Poll<Result<Infallible, InnerError>> {
457        self.process_waiting_to_close();
458
459        self.waiting_to_close_waker = ctx.waker().clone();
460
461        loop {
462            if let Poll::Ready(e) = self.transport.poll_send_messages(ctx) {
463                for state in std::mem::take(&mut self.socket_read_states).into_values() {
464                    state.wakers.into_iter().for_each(Waker::wake);
465                }
466                for (_, state) in self.channel_read_states.drain() {
467                    state.wakers.into_iter().for_each(Waker::wake);
468                }
469                return Poll::Ready(Err(e));
470            }
471            let Poll::Ready(result) = self.transport.poll_next(ctx) else {
472                return Poll::Pending;
473            };
474            let data = result?;
475            let (header, data) = match fidl_message::decode_transaction_header(&data) {
476                Ok(x) => x,
477                Err(e) => {
478                    self.transport = Transport::Error(InnerError::Protocol(e));
479                    continue;
480                }
481            };
482
483            let Some(tx_id) = NonZeroU32::new(header.tx_id) else {
484                match self.process_event(header, data) {
485                    Ok(wakers) => self.wakers_to_wake.extend(wakers),
486                    Err(e) => self.transport = Transport::Error(e),
487                }
488                continue;
489            };
490
491            let tx = self.transactions.remove(&tx_id).ok_or(::fidl::Error::InvalidResponseTxid)?;
492            match tx.handle(self, Ok((header, data))) {
493                Ok(x) => x,
494                Err(e) => {
495                    self.transport = Transport::Error(InnerError::Protocol(e));
496                    continue;
497                }
498            }
499        }
500    }
501
502    /// Process an incoming message that arose from an event rather than a transaction reply.
503    fn process_event(
504        &mut self,
505        header: TransactionHeader,
506        data: &[u8],
507    ) -> Result<Vec<Waker>, InnerError> {
508        match header.ordinal {
509            ordinals::ON_SOCKET_STREAMING_DATA => {
510                let msg = fidl_message::decode_message::<proto::SocketOnSocketStreamingDataRequest>(
511                    header, data,
512                )?;
513                let o =
514                    self.socket_read_states.entry(msg.handle).or_insert_with(|| SocketReadState {
515                        wakers: Vec::new(),
516                        queued: VecDeque::new(),
517                        is_streaming: false,
518                        read_request_pending: false,
519                    });
520                match msg.socket_message {
521                    proto::SocketMessage::Data(data) => Ok(o.handle_incoming_message(Ok(data))),
522                    proto::SocketMessage::Stopped(proto::AioStopped { error }) => {
523                        let ret = if let Some(error) = error {
524                            o.handle_incoming_message(Err(Error::FDomain(*error)))
525                        } else {
526                            Vec::new()
527                        };
528                        o.is_streaming = false;
529                        Ok(ret)
530                    }
531                    _ => Err(InnerError::ProtocolStreamEventIncompatible),
532                }
533            }
534            ordinals::ON_CHANNEL_STREAMING_DATA => {
535                let msg = fidl_message::decode_message::<
536                    proto::ChannelOnChannelStreamingDataRequest,
537                >(header, data)?;
538                let o = self.channel_read_states.entry(msg.handle).or_insert_with(|| {
539                    ChannelReadState {
540                        wakers: Vec::new(),
541                        queued: VecDeque::new(),
542                        is_streaming: false,
543                        read_request_pending: false,
544                    }
545                });
546                match msg.channel_sent {
547                    proto::ChannelSent::Message(data) => Ok(o.handle_incoming_message(Ok(data))),
548                    proto::ChannelSent::Stopped(proto::AioStopped { error }) => {
549                        let ret = if let Some(error) = error {
550                            o.handle_incoming_message(Err(Error::FDomain(*error)))
551                        } else {
552                            Vec::new()
553                        };
554                        o.is_streaming = false;
555                        Ok(ret)
556                    }
557                    _ => Err(InnerError::ProtocolStreamEventIncompatible),
558                }
559            }
560            _ => Err(::fidl::Error::UnknownOrdinal {
561                ordinal: header.ordinal,
562                protocol_name:
563                    <proto::FDomainMarker as ::fidl::endpoints::ProtocolMarker>::DEBUG_NAME,
564            }
565            .into()),
566        }
567    }
568
569    /// Polls the underlying transport to ensure any incoming or outgoing
570    /// messages are processed as far as possible. If a failure occurs, puts the
571    /// transport into an error state and fails all pending transactions.
572    fn poll_transport(&mut self, ctx: &mut Context<'_>) -> Poll<()> {
573        if let Poll::Ready(Err(e)) = self.try_poll_transport(ctx) {
574            for (_, v) in std::mem::take(&mut self.transactions) {
575                let _ = v.handle(self, Err(e.clone()));
576            }
577
578            Poll::Ready(())
579        } else {
580            Poll::Pending
581        }
582    }
583
584    /// Handles the response to a `SocketRead` protocol message.
585    pub(crate) fn handle_socket_read_response(
586        &mut self,
587        msg: Result<proto::SocketData, Error>,
588        id: proto::HandleId,
589    ) {
590        let state = self.socket_read_states.entry(id).or_insert_with(|| SocketReadState {
591            wakers: Vec::new(),
592            queued: VecDeque::new(),
593            is_streaming: false,
594            read_request_pending: false,
595        });
596        let wakers = state.handle_incoming_message(msg);
597        self.wakers_to_wake.extend(wakers);
598        state.read_request_pending = false;
599    }
600
601    /// Handles the response to a `ChannelRead` protocol message.
602    pub(crate) fn handle_channel_read_response(
603        &mut self,
604        msg: Result<proto::ChannelMessage, Error>,
605        id: proto::HandleId,
606    ) {
607        let state = self.channel_read_states.entry(id).or_insert_with(|| ChannelReadState {
608            wakers: Vec::new(),
609            queued: VecDeque::new(),
610            is_streaming: false,
611            read_request_pending: false,
612        });
613        let wakers = state.handle_incoming_message(msg);
614        self.wakers_to_wake.extend(wakers);
615        state.read_request_pending = false;
616    }
617}
618
619impl Drop for ClientInner {
620    fn drop(&mut self) {
621        let responders = self.transactions.drain().map(|x| x.1).collect::<Vec<_>>();
622        for responder in responders {
623            let _ = responder.handle(self, Err(InnerError::Transport(None)));
624        }
625        for state in self.channel_read_states.values_mut() {
626            state.wakers.drain(..).for_each(Waker::wake);
627        }
628        for state in self.socket_read_states.values_mut() {
629            state.wakers.drain(..).for_each(Waker::wake);
630        }
631        self.waiting_to_close_waker.wake_by_ref();
632        self.wakers_to_wake.drain(..).for_each(Waker::wake);
633    }
634}
635
636/// Represents a connection to an FDomain.
637///
638/// The client is constructed by passing it a transport object which represents
639/// the raw connection to the remote FDomain. The `Client` wrapper then allows
640/// us to construct and use handles which behave similarly to their counterparts
641/// on a Fuchsia device.
642pub struct Client(pub(crate) Mutex<ClientInner>);
643
644impl std::fmt::Debug for Client {
645    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
646        let inner = self.0.lock();
647        match &inner.transport {
648            Transport::Transport(transport, ..) if transport.has_debug_fmt() => {
649                write!(f, "Client(")?;
650                transport.debug_fmt(f)?;
651                write!(f, ")")
652            }
653            Transport::Error(error) => {
654                let error = Error::from(error.clone());
655                write!(f, "Client(Failed: {error})")
656            }
657            _ => f.debug_tuple("Client").field(&"<transport>").finish(),
658        }
659    }
660}
661
662/// A client which is always disconnected. Handles that lose their clients
663/// connect to this client instead, which always returns a "Client Lost"
664/// transport failure.
665pub(crate) static DEAD_CLIENT: LazyLock<Arc<Client>> = LazyLock::new(|| {
666    Arc::new(Client(Mutex::new(ClientInner {
667        transport: Transport::Error(InnerError::Transport(None)),
668        transactions: HashMap::new(),
669        channel_read_states: HashMap::new(),
670        socket_read_states: HashMap::new(),
671        next_tx_id: 1,
672        waiting_to_close: Vec::new(),
673        waiting_to_close_waker: std::task::Waker::noop().clone(),
674        wakers_to_wake: Vec::new(),
675    })))
676});
677
678impl Client {
679    /// Create a new FDomain client. The `transport` argument should contain the
680    /// established connection to the target, ready to communicate the FDomain
681    /// protocol.
682    ///
683    /// The second return item is a future that must be polled to keep
684    /// transactions running.
685    pub fn new(
686        transport: impl FDomainTransport + 'static,
687    ) -> (Arc<Self>, impl Future<Output = ()> + Send + 'static) {
688        let ret = Arc::new(Client(Mutex::new(ClientInner {
689            transport: Transport::Transport(Box::pin(transport), VecDeque::new(), Vec::new()),
690            transactions: HashMap::new(),
691            socket_read_states: HashMap::new(),
692            channel_read_states: HashMap::new(),
693            next_tx_id: 1,
694            waiting_to_close: Vec::new(),
695            waiting_to_close_waker: std::task::Waker::noop().clone(),
696            wakers_to_wake: Vec::new(),
697        })));
698
699        let client_weak = Arc::downgrade(&ret);
700        let fut = futures::future::poll_fn(move |ctx| {
701            let Some(client) = client_weak.upgrade() else {
702                return Poll::Ready(());
703            };
704
705            let (ret, deferred_wakers) = {
706                let mut inner = client.0.lock();
707                let ret = inner.poll_transport(ctx);
708                let deferred_wakers = std::mem::replace(&mut inner.wakers_to_wake, Vec::new());
709                (ret, deferred_wakers)
710            };
711            deferred_wakers.into_iter().for_each(Waker::wake);
712            ret
713        });
714
715        (ret, fut)
716    }
717
718    /// Get the namespace for the connected FDomain. Calling this more than once is an error.
719    pub async fn namespace(self: &Arc<Self>) -> Result<Channel, Error> {
720        let new_handle = self.new_hid();
721        self.transaction(
722            ordinals::GET_NAMESPACE,
723            proto::FDomainGetNamespaceRequest { new_handle },
724            Responder::Namespace,
725        )
726        .await?;
727        Ok(Channel(Handle { id: new_handle.id, client: Arc::downgrade(self) }))
728    }
729
730    /// Create a new channel in the connected FDomain.
731    pub fn create_channel(self: &Arc<Self>) -> (Channel, Channel) {
732        let id_a = self.new_hid();
733        let id_b = self.new_hid();
734        let fut = self.transaction(
735            ordinals::CREATE_CHANNEL,
736            proto::ChannelCreateChannelRequest { handles: [id_a, id_b] },
737            Responder::CreateChannel,
738        );
739
740        fuchsia_async::Task::spawn(async move {
741            if let Err(e) = fut.await {
742                log::debug!("FDomain channel creation failed: {e}");
743            }
744        })
745        .detach();
746
747        (
748            Channel(Handle { id: id_a.id, client: Arc::downgrade(self) }),
749            Channel(Handle { id: id_b.id, client: Arc::downgrade(self) }),
750        )
751    }
752
753    /// Creates client and server endpoints connected to by a channel.
754    pub fn create_endpoints<F: crate::fidl::ProtocolMarker>(
755        self: &Arc<Self>,
756    ) -> (crate::fidl::ClientEnd<F>, crate::fidl::ServerEnd<F>) {
757        let (client, server) = self.create_channel();
758        let client_end = crate::fidl::ClientEnd::<F>::new(client);
759        let server_end = crate::fidl::ServerEnd::new(server);
760        (client_end, server_end)
761    }
762
763    /// Creates a client proxy and a server endpoint connected by a channel.
764    pub fn create_proxy<F: crate::fidl::ProtocolMarker>(
765        self: &Arc<Self>,
766    ) -> (F::Proxy, crate::fidl::ServerEnd<F>) {
767        let (client_end, server_end) = self.create_endpoints::<F>();
768        (client_end.into_proxy(), server_end)
769    }
770
771    /// Creates a client proxy and a server request stream connected by a channel.
772    pub fn create_proxy_and_stream<F: crate::fidl::ProtocolMarker>(
773        self: &Arc<Self>,
774    ) -> (F::Proxy, F::RequestStream) {
775        let (client_end, server_end) = self.create_endpoints::<F>();
776        (client_end.into_proxy(), server_end.into_stream())
777    }
778
779    /// Creates a client end and a server request stream connected by a channel.
780    pub fn create_request_stream<F: crate::fidl::ProtocolMarker>(
781        self: &Arc<Self>,
782    ) -> (crate::fidl::ClientEnd<F>, F::RequestStream) {
783        let (client_end, server_end) = self.create_endpoints::<F>();
784        (client_end, server_end.into_stream())
785    }
786
787    /// Create a new socket in the connected FDomain.
788    fn create_socket(self: &Arc<Self>, options: proto::SocketType) -> (Socket, Socket) {
789        let id_a = self.new_hid();
790        let id_b = self.new_hid();
791        let fut = self.transaction(
792            ordinals::CREATE_SOCKET,
793            proto::SocketCreateSocketRequest { handles: [id_a, id_b], options },
794            Responder::CreateSocket,
795        );
796
797        fuchsia_async::Task::spawn(async move {
798            if let Err(e) = fut.await {
799                log::debug!("FDomain socket creation failed: {e}");
800            }
801        })
802        .detach();
803
804        (
805            Socket(Handle { id: id_a.id, client: Arc::downgrade(self) }),
806            Socket(Handle { id: id_b.id, client: Arc::downgrade(self) }),
807        )
808    }
809
810    /// Create a new streaming socket in the connected FDomain.
811    pub fn create_stream_socket(self: &Arc<Self>) -> (Socket, Socket) {
812        self.create_socket(proto::SocketType::Stream)
813    }
814
815    /// Create a new datagram socket in the connected FDomain.
816    pub fn create_datagram_socket(self: &Arc<Self>) -> (Socket, Socket) {
817        self.create_socket(proto::SocketType::Datagram)
818    }
819
820    /// Create a new event pair in the connected FDomain.
821    pub fn create_event_pair(self: &Arc<Self>) -> (EventPair, EventPair) {
822        let id_a = self.new_hid();
823        let id_b = self.new_hid();
824        let fut = self.transaction(
825            ordinals::CREATE_EVENT_PAIR,
826            proto::EventPairCreateEventPairRequest { handles: [id_a, id_b] },
827            Responder::CreateEventPair,
828        );
829
830        fuchsia_async::Task::spawn(async move {
831            if let Err(e) = fut.await {
832                log::debug!("FDomain event pair creation failed: {e}");
833            }
834        })
835        .detach();
836
837        (
838            EventPair(Handle { id: id_a.id, client: Arc::downgrade(self) }),
839            EventPair(Handle { id: id_b.id, client: Arc::downgrade(self) }),
840        )
841    }
842
843    /// Create a new event handle in the connected FDomain.
844    pub fn create_event(self: &Arc<Self>) -> Event {
845        let id = self.new_hid();
846        let fut = self.transaction(
847            ordinals::CREATE_EVENT,
848            proto::EventCreateEventRequest { handle: id },
849            Responder::CreateEvent,
850        );
851
852        fuchsia_async::Task::spawn(async move {
853            if let Err(e) = fut.await {
854                log::debug!("FDomain event creation failed: {e}");
855            }
856        })
857        .detach();
858
859        Event(Handle { id: id.id, client: Arc::downgrade(self) })
860    }
861
862    /// Allocate a new HID, which should be suitable for use with the connected FDomain.
863    pub(crate) fn new_hid(&self) -> proto::NewHandleId {
864        // TODO: On the target side we have to keep a table of these which means
865        // we can automatically detect collisions in the random value. On the
866        // client side we'd have to add a whole data structure just for that
867        // purpose. Should we?
868        proto::NewHandleId { id: rand::random::<u32>() >> 1 }
869    }
870
871    /// Create a future which sends a FIDL message to the connected FDomain and
872    /// waits for a response.
873    ///
874    /// Calling this method queues the transaction synchronously. Awaiting is
875    /// only necessary to wait for the response.
876    pub(crate) fn transaction<S: fidl_message::Body, R: 'static, F>(
877        self: &Arc<Self>,
878        ordinal: u64,
879        request: S,
880        f: F,
881    ) -> impl Future<Output = Result<R, Error>> + 'static + use<S, R, F>
882    where
883        F: Fn(OneshotSender<Result<R, Error>>) -> Responder,
884    {
885        let mut inner = self.0.lock();
886
887        let (sender, receiver) = futures::channel::oneshot::channel();
888        inner.request(ordinal, request, f(sender));
889        receiver.map(|x| x.expect("Oneshot went away without reply!"))
890    }
891
892    /// Start getting streaming events for socket reads.
893    pub(crate) fn start_socket_streaming(&self, id: proto::HandleId) -> Result<(), Error> {
894        let mut inner = self.0.lock();
895        if let Some(e) = inner.transport.error() {
896            return Err(e.into());
897        }
898
899        let state = inner.socket_read_states.entry(id).or_insert_with(|| SocketReadState {
900            wakers: Vec::new(),
901            queued: VecDeque::new(),
902            is_streaming: false,
903            read_request_pending: false,
904        });
905
906        assert!(!state.is_streaming, "Initiated streaming twice!");
907        state.is_streaming = true;
908
909        inner.request(
910            ordinals::READ_SOCKET_STREAMING_START,
911            proto::SocketReadSocketStreamingStartRequest { handle: id },
912            Responder::Ignore,
913        );
914        Ok(())
915    }
916
917    /// Stop getting streaming events for socket reads. Doesn't return errors
918    /// because it's exclusively called in destructors where we have nothing to
919    /// do with them.
920    pub(crate) fn stop_socket_streaming(&self, id: proto::HandleId) {
921        let mut inner = self.0.lock();
922        if let Some(state) = inner.socket_read_states.get_mut(&id) {
923            if state.is_streaming {
924                state.is_streaming = false;
925                // TODO: Log?
926                let _ = inner.request(
927                    ordinals::READ_SOCKET_STREAMING_STOP,
928                    proto::ChannelReadChannelStreamingStopRequest { handle: id },
929                    Responder::Ignore,
930                );
931            }
932        }
933    }
934
935    /// Start getting streaming events for socket reads.
936    pub(crate) fn start_channel_streaming(&self, id: proto::HandleId) -> Result<(), Error> {
937        let mut inner = self.0.lock();
938        if let Some(e) = inner.transport.error() {
939            return Err(e.into());
940        }
941        let state = inner.channel_read_states.entry(id).or_insert_with(|| ChannelReadState {
942            wakers: Vec::new(),
943            queued: VecDeque::new(),
944            is_streaming: false,
945            read_request_pending: false,
946        });
947
948        assert!(!state.is_streaming, "Initiated streaming twice!");
949        state.is_streaming = true;
950
951        inner.request(
952            ordinals::READ_CHANNEL_STREAMING_START,
953            proto::ChannelReadChannelStreamingStartRequest { handle: id },
954            Responder::Ignore,
955        );
956
957        Ok(())
958    }
959
960    /// Stop getting streaming events for socket reads. Doesn't return errors
961    /// because it's exclusively called in destructors where we have nothing to
962    /// do with them.
963    pub(crate) fn stop_channel_streaming(&self, id: proto::HandleId) {
964        let mut inner = self.0.lock();
965        if let Some(state) = inner.channel_read_states.get_mut(&id) {
966            if state.is_streaming {
967                state.is_streaming = false;
968                // TODO: Log?
969                let _ = inner.request(
970                    ordinals::READ_CHANNEL_STREAMING_STOP,
971                    proto::ChannelReadChannelStreamingStopRequest { handle: id },
972                    Responder::Ignore,
973                );
974            }
975        }
976    }
977
978    /// Execute a read from a channel.
979    pub(crate) fn poll_socket(
980        &self,
981        id: proto::HandleId,
982        ctx: &mut Context<'_>,
983        out: &mut [u8],
984    ) -> Poll<Result<usize, Error>> {
985        let mut inner = self.0.lock();
986        if let Some(error) = inner.transport.error() {
987            return Poll::Ready(Err(error.into()));
988        }
989
990        let state = inner.socket_read_states.entry(id).or_insert_with(|| SocketReadState {
991            wakers: Vec::new(),
992            queued: VecDeque::new(),
993            is_streaming: false,
994            read_request_pending: false,
995        });
996
997        if let Some(got) = state.queued.front_mut() {
998            match got.as_mut() {
999                Ok(data) => {
1000                    let read_size = std::cmp::min(data.data.len(), out.len());
1001                    out[..read_size].copy_from_slice(&data.data[..read_size]);
1002
1003                    if data.data.len() > read_size && !data.is_datagram {
1004                        let _ = data.data.drain(..read_size);
1005                    } else {
1006                        let _ = state.queued.pop_front();
1007                    }
1008
1009                    return Poll::Ready(Ok(read_size));
1010                }
1011                Err(_) => {
1012                    let err = state.queued.pop_front().unwrap().unwrap_err();
1013                    return Poll::Ready(Err(err));
1014                }
1015            }
1016        } else if !state.wakers.iter().any(|x| ctx.waker().will_wake(x)) {
1017            state.wakers.push(ctx.waker().clone());
1018        }
1019
1020        if !state.read_request_pending && !state.is_streaming {
1021            inner.request(
1022                ordinals::READ_SOCKET,
1023                proto::SocketReadSocketRequest { handle: id, max_bytes: out.len() as u64 },
1024                Responder::ReadSocket(id),
1025            );
1026        }
1027
1028        Poll::Pending
1029    }
1030
1031    /// Execute a read from a channel.
1032    pub(crate) fn poll_channel(
1033        &self,
1034        id: proto::HandleId,
1035        ctx: &mut Context<'_>,
1036        for_stream: bool,
1037    ) -> Poll<Option<Result<proto::ChannelMessage, Error>>> {
1038        let mut inner = self.0.lock();
1039        if let Some(error) = inner.transport.error() {
1040            return Poll::Ready(Some(Err(error.into())));
1041        }
1042
1043        let state = inner.channel_read_states.entry(id).or_insert_with(|| ChannelReadState {
1044            wakers: Vec::new(),
1045            queued: VecDeque::new(),
1046            is_streaming: false,
1047            read_request_pending: false,
1048        });
1049
1050        if let Some(got) = state.queued.pop_front() {
1051            return Poll::Ready(Some(got));
1052        } else if for_stream && !state.is_streaming {
1053            return Poll::Ready(None);
1054        } else if !state.wakers.iter().any(|x| ctx.waker().will_wake(x)) {
1055            state.wakers.push(ctx.waker().clone());
1056        }
1057
1058        if !state.read_request_pending && !state.is_streaming {
1059            inner.request(
1060                ordinals::READ_CHANNEL,
1061                proto::ChannelReadChannelRequest { handle: id },
1062                Responder::ReadChannel(id),
1063            );
1064        }
1065
1066        Poll::Pending
1067    }
1068
1069    /// Check whether this channel is streaming
1070    pub(crate) fn channel_is_streaming(&self, id: proto::HandleId) -> bool {
1071        let inner = self.0.lock();
1072        let Some(state) = inner.channel_read_states.get(&id) else {
1073            return false;
1074        };
1075        state.is_streaming
1076    }
1077
1078    /// Check that all the given handles are safe to transfer through a channel
1079    /// e.g. that there's no chance of in-flight reads getting dropped.
1080    pub(crate) fn clear_handles_for_transfer(&self, handles: &proto::Handles) {
1081        let inner = self.0.lock();
1082        match handles {
1083            proto::Handles::Handles(handles) => {
1084                for handle in handles {
1085                    assert!(
1086                        !(inner.channel_read_states.contains_key(handle)
1087                            || inner.socket_read_states.contains_key(handle)),
1088                        "Tried to transfer handle after reading"
1089                    );
1090                }
1091            }
1092            proto::Handles::Dispositions(dispositions) => {
1093                for disposition in dispositions {
1094                    match &disposition.handle {
1095                        proto::HandleOp::Move_(handle) => assert!(
1096                            !(inner.channel_read_states.contains_key(handle)
1097                                || inner.socket_read_states.contains_key(handle)),
1098                            "Tried to transfer handle after reading"
1099                        ),
1100                        // Pretty sure this should be fine regardless of read state.
1101                        proto::HandleOp::Duplicate(_) => (),
1102                    }
1103                }
1104            }
1105        }
1106    }
1107}
fdomain_client/lib.rs

fdomain_client/
lib.rs
