fdomain_uart_driver_lib/
coordinator.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Multiplexed FDomain channel coordinator and RCS socket bridge tasks.
6
7use crate::error::Result;
8use crate::receiver::SerialEvent;
9use crate::sender::SenderMessage;
10use fidl_fuchsia_developer_remotecontrol_connector::ConnectorProxy;
11use futures::channel::mpsc;
12use futures::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf};
13use futures::prelude::*;
14use log::{error, warn};
15use std::collections::{BTreeMap, HashMap, VecDeque};
16use std::pin::Pin;
17use std::task::{Context, Poll};
18use uart_fpl::{CONTROL_CHANNEL_ID, MAX_PAYLOAD_SIZE};
19
20/// Default bounded mpsc channel capacity for internal control and frame queues.
21pub const DEFAULT_CHANNEL_CAPACITY: usize = 64;
22
23/// Default bounded mpsc channel capacity for multiplexed client data queues.
24pub const DEFAULT_CLIENT_DATA_CAPACITY: usize = 64;
25
26/// Maximum queue depth for per-channel asynchronous socket writers.
27const CHANNEL_WRITER_QUEUE_CAPACITY: usize = 256;
28
29/// Maximum number of concurrently open or pending FDomain channels.
30const MAX_ACTIVE_CHANNELS: usize = 256;
31
32/// Maximum buffer capacity (in bytes) staged for a single channel before socket connection.
33const MAX_CHANNEL_STAGING_BYTES: usize = 2 * 1024 * 1024;
34
35/// Events sent by asynchronous coordinator worker tasks to the main loop.
36#[derive(Debug)]
37enum InternalEvent {
38    /// Asynchronous toolbox socket registration succeeded for a channel.
39    Registered { channel_id: u16, generation: u64, socket: fuchsia_async::Socket },
40    /// Asynchronous toolbox socket registration failed for a channel.
41    RegistrationFailed { channel_id: u16, generation: u64, error: fidl::Error },
42    /// A channel writer task finished flushing its queue (`error: None`) or encountered an error.
43    WriterDone { channel_id: u16, generation: u64, error: Option<std::io::Error> },
44}
45
46/// Events sent by per-channel RCS socket readers to the coordinator.
47#[derive(Debug)]
48enum ClientEvent {
49    /// Ingress data read from the local RCS socket to be sent over serial.
50    Data { channel_id: u16, generation: u64, payload: Vec<u8> },
51    /// The local RCS socket reached EOF (`error: None`) or encountered a read error.
52    Closed { channel_id: u16, generation: u64, error: Option<std::io::Error> },
53}
54
55/// Lifecycle state of an active or pending FDomain channel.
56enum ChannelState {
57    /// Channel is waiting for `fdomain_toolbox_socket` FIDL registration to complete.
58    Pending {
59        generation: u64,
60        buffer: Vec<Vec<u8>>,
61        pending_bytes: usize,
62        _registration_task: fuchsia_async::Task<()>,
63    },
64    /// Channel has an open, active bidirectional RCS socket bridge.
65    Connected {
66        generation: u64,
67        sender: mpsc::Sender<Vec<u8>>,
68        pending_incoming: VecDeque<Vec<u8>>,
69        pending_bytes: usize,
70        _reader_task: fuchsia_async::Task<()>,
71        _writer_task: fuchsia_async::Task<()>,
72    },
73}
74
75/// State machine for multiplexing FDomain channels and scheduling egress messages.
76struct CoordinatorState {
77    active: HashMap<u16, ChannelState>,
78    closing_pending: HashMap<(u16, u64), (Vec<Vec<u8>>, fuchsia_async::Task<()>)>,
79    closing_writers: HashMap<(u16, u64), fuchsia_async::Task<()>>,
80    outgoing: BTreeMap<u16, VecDeque<SenderMessage>>,
81    last_channel_id: Option<u16>,
82    next_generation: u64,
83    client_tx: mpsc::Sender<ClientEvent>,
84    internal_tx: mpsc::Sender<InternalEvent>,
85}
86
87/// Reads data from an active RCS socket and forwards it as [`ClientEvent`]s to the coordinator.
88///
89/// When EOF (`Ok(0)`) or a socket read error (`Err(e)`) occurs, a [`ClientEvent::Closed`] is sent
90/// carrying `error: Option<std::io::Error>` so [`CoordinatorState`] can distinguish normal socket
91/// closure from transport errors on active generations.
92async fn channel_reader_task(
93    channel_id: u16,
94    generation: u64,
95    mut reader: ReadHalf<fuchsia_async::Socket>,
96    mut client_data_tx: mpsc::Sender<ClientEvent>,
97) {
98    let mut buf = [0u8; MAX_PAYLOAD_SIZE];
99    loop {
100        match reader.read(&mut buf).await {
101            Ok(0) => {
102                let _ = client_data_tx
103                    .send(ClientEvent::Closed { channel_id, generation, error: None })
104                    .await;
105                break;
106            }
107            Ok(n) => {
108                let event =
109                    ClientEvent::Data { channel_id, generation, payload: buf[..n].to_vec() };
110                if let Err(e) = client_data_tx.send(event).await {
111                    error!("Failed to forward RCS data for channel {}: {:?}", channel_id, e);
112                    break;
113                }
114            }
115            Err(e) => {
116                let _ = client_data_tx
117                    .send(ClientEvent::Closed { channel_id, generation, error: Some(e) })
118                    .await;
119                break;
120            }
121        }
122    }
123}
124
125/// Writes buffered and incoming channel payloads from the coordinator to the active RCS socket.
126///
127/// Once all buffered and queued data has been written and flushed (or if a write fails), an
128/// [`InternalEvent::WriterDone`] notification is sent back to the coordinator along with any
129/// write/flush error encountered.
130async fn channel_writer_task(
131    channel_id: u16,
132    generation: u64,
133    mut writer: WriteHalf<fuchsia_async::Socket>,
134    mut receiver: mpsc::Receiver<Vec<u8>>,
135    buffer: Vec<Vec<u8>>,
136    mut internal_tx: mpsc::Sender<InternalEvent>,
137) {
138    async fn write_and_flush(
139        writer: &mut WriteHalf<fuchsia_async::Socket>,
140        data: &[u8],
141    ) -> std::io::Result<()> {
142        writer.write_all(data).await?;
143        writer.flush().await
144    }
145
146    let mut write_error = None;
147    for data in buffer {
148        if let Err(e) = write_and_flush(&mut writer, &data).await {
149            write_error = Some(e);
150            break;
151        }
152    }
153    while write_error.is_none()
154        && let Some(data) = receiver.next().await
155    {
156        if let Err(e) = write_and_flush(&mut writer, &data).await {
157            write_error = Some(e);
158            break;
159        }
160    }
161    let _ = internal_tx
162        .send(InternalEvent::WriterDone { channel_id, generation, error: write_error })
163        .await;
164}
165
166/// Attempts to send an incoming chunk directly to the writer channel, or stages it in
167/// `pending_incoming` if the writer channel is temporarily full.
168fn stage_or_send_connected(
169    channel_id: u16,
170    data: Vec<u8>,
171    sender: &mut mpsc::Sender<Vec<u8>>,
172    pending_incoming: &mut VecDeque<Vec<u8>>,
173    pending_bytes: &mut usize,
174) -> std::result::Result<(), String> {
175    let to_stage = if pending_incoming.is_empty() {
176        match sender.try_send(data) {
177            Ok(()) => return Ok(()),
178            Err(e) if e.is_disconnected() => {
179                let reason = format!("Channel {channel_id} writer disconnected");
180                error!("{reason}, dropping channel");
181                return Err(reason);
182            }
183            Err(e) => e.into_inner(),
184        }
185    } else {
186        data
187    };
188    if *pending_bytes + to_stage.len() > MAX_CHANNEL_STAGING_BYTES {
189        let reason = format!(
190            "Channel {channel_id} exceeded max staging buffer ({MAX_CHANNEL_STAGING_BYTES} bytes)"
191        );
192        error!("{reason}, dropping channel");
193        return Err(reason);
194    }
195    *pending_bytes += to_stage.len();
196    pending_incoming.push_back(to_stage);
197    Ok(())
198}
199
200/// Stages an incoming serial payload in the pending channel buffer before socket connection.
201fn stage_pending_data(
202    channel_id: u16,
203    data: Vec<u8>,
204    buffer: &mut Vec<Vec<u8>>,
205    pending_bytes: &mut usize,
206) -> std::result::Result<(), String> {
207    if *pending_bytes + data.len() > MAX_CHANNEL_STAGING_BYTES {
208        let reason = format!(
209            "Pending channel {channel_id} exceeded max staging buffer ({MAX_CHANNEL_STAGING_BYTES} bytes)"
210        );
211        error!("{reason}, dropping channel");
212        return Err(reason);
213    }
214    *pending_bytes += data.len();
215    buffer.push(data);
216    Ok(())
217}
218
219/// Spawns an asynchronous task to register a channel socket with the RCS connector.
220fn spawn_channel_registration(
221    channel_id: u16,
222    generation: u64,
223    rcs_connector: ConnectorProxy,
224    mut internal_tx: mpsc::Sender<InternalEvent>,
225) -> fuchsia_async::Task<()> {
226    fuchsia_async::Task::spawn(async move {
227        let (local_socket, remote_socket) = fidl::Socket::create_stream();
228        if let Err(error) = rcs_connector.fdomain_toolbox_socket(remote_socket).await {
229            let _ = internal_tx
230                .send(InternalEvent::RegistrationFailed { channel_id, generation, error })
231                .await;
232            return;
233        }
234        let socket = fuchsia_async::Socket::from_socket(local_socket);
235        let _ =
236            internal_tx.send(InternalEvent::Registered { channel_id, generation, socket }).await;
237    })
238}
239
240impl CoordinatorState {
241    /// Creates a new [`CoordinatorState`].
242    fn new(client_tx: mpsc::Sender<ClientEvent>, internal_tx: mpsc::Sender<InternalEvent>) -> Self {
243        Self {
244            active: HashMap::new(),
245            closing_pending: HashMap::new(),
246            closing_writers: HashMap::new(),
247            outgoing: BTreeMap::new(),
248            last_channel_id: None,
249            next_generation: 1,
250            client_tx,
251            internal_tx,
252        }
253    }
254
255    /// Allocates a new monotonically increasing generation identifier for a channel session.
256    fn alloc_generation(&mut self) -> u64 {
257        let generation = self.next_generation;
258        self.next_generation = self.next_generation.wrapping_add(1);
259        generation
260    }
261
262    /// Enqueues a [`SenderMessage::Close`] message into the outgoing serial queue for `channel_id`.
263    fn enqueue_close(&mut self, channel_id: u16, reason: Option<String>) {
264        self.outgoing
265            .entry(channel_id)
266            .or_default()
267            .push_back(SenderMessage::Close { channel_id, reason });
268    }
269
270    /// Synchronously drains staged incoming chunks from `pending_incoming` into the connected
271    /// writer sender channels using non-blocking `try_send`.
272    ///
273    /// If a channel writer channel is disconnected, the channel is marked as disconnected and
274    /// queued for closure.
275    fn drain_pending_incoming(&mut self) {
276        let mut disconnected_channels = Vec::new();
277        for (&channel_id, state) in self.active.iter_mut() {
278            if let ChannelState::Connected { sender, pending_incoming, pending_bytes, .. } = state {
279                while let Some(front) = pending_incoming.pop_front() {
280                    let len = front.len();
281                    match sender.try_send(front) {
282                        Ok(()) => *pending_bytes = pending_bytes.saturating_sub(len),
283                        Err(e) if e.is_disconnected() => {
284                            disconnected_channels.push(channel_id);
285                            break;
286                        }
287                        Err(e) => {
288                            pending_incoming.push_front(e.into_inner());
289                            break;
290                        }
291                    }
292                }
293            }
294        }
295        for channel_id in disconnected_channels {
296            if self.active.remove(&channel_id).is_some() {
297                self.enqueue_close(
298                    channel_id,
299                    Some(format!("Channel {channel_id} writer disconnected")),
300                );
301            }
302        }
303    }
304
305    /// Asynchronously polls readiness of the connected channel writer sinks that have staged
306    /// incoming data in `pending_incoming`.
307    ///
308    /// Returns `Poll::Ready(())` if any staged data was successfully sent into a writer sink,
309    /// waking the event loop to continue making progress.
310    fn poll_pending_incoming(&mut self, cx: &mut Context<'_>) -> Poll<()> {
311        let mut made_progress = false;
312        let mut disconnected_channels = Vec::new();
313        for (&channel_id, state) in self.active.iter_mut() {
314            if let ChannelState::Connected { sender, pending_incoming, pending_bytes, .. } = state {
315                while !pending_incoming.is_empty() {
316                    match futures::sink::Sink::poll_ready(Pin::new(&mut *sender), cx) {
317                        Poll::Ready(Ok(())) => {
318                            let front = pending_incoming.pop_front().unwrap();
319                            let len = front.len();
320                            if futures::sink::Sink::start_send(Pin::new(&mut *sender), front)
321                                .is_err()
322                            {
323                                disconnected_channels.push(channel_id);
324                                break;
325                            }
326                            *pending_bytes = pending_bytes.saturating_sub(len);
327                            made_progress = true;
328                        }
329                        Poll::Ready(Err(_)) => {
330                            disconnected_channels.push(channel_id);
331                            break;
332                        }
333                        Poll::Pending => break,
334                    }
335                }
336            }
337        }
338        for channel_id in disconnected_channels {
339            if self.active.remove(&channel_id).is_some() {
340                self.enqueue_close(
341                    channel_id,
342                    Some(format!("Channel {channel_id} writer disconnected")),
343                );
344            }
345        }
346        if made_progress { Poll::Ready(()) } else { Poll::Pending }
347    }
348
349    /// Binds an established RCS socket to `channel_id`, spawning reader and writer tasks.
350    ///
351    /// If `closing` is true, the socket is bound in a closing/draining state: the writer task is
352    /// spawned to flush `buffer` before closing, without spawning a reader task.
353    fn connect_socket(
354        &mut self,
355        channel_id: u16,
356        generation: u64,
357        socket: fuchsia_async::Socket,
358        buffer: Vec<Vec<u8>>,
359        closing: bool,
360    ) {
361        let (reader, writer) = socket.split();
362        let (mut writer_tx, writer_rx) = mpsc::channel(CHANNEL_WRITER_QUEUE_CAPACITY);
363        let writer_task = fuchsia_async::Task::spawn(channel_writer_task(
364            channel_id,
365            generation,
366            writer,
367            writer_rx,
368            buffer,
369            self.internal_tx.clone(),
370        ));
371        if closing {
372            writer_tx.close_channel();
373            self.closing_writers.insert((channel_id, generation), writer_task);
374            return;
375        }
376        let reader_task = fuchsia_async::Task::spawn(channel_reader_task(
377            channel_id,
378            generation,
379            reader,
380            self.client_tx.clone(),
381        ));
382        self.active.insert(
383            channel_id,
384            ChannelState::Connected {
385                generation,
386                sender: writer_tx,
387                pending_incoming: VecDeque::new(),
388                pending_bytes: 0,
389                _reader_task: reader_task,
390                _writer_task: writer_task,
391            },
392        );
393    }
394
395    /// Handles a successful socket registration event from a background worker task.
396    ///
397    /// Connects the newly opened RCS socket if the channel generation matches and the channel
398    /// remains active or is in the process of closing.
399    fn handle_registered_event(
400        &mut self,
401        channel_id: u16,
402        generation: u64,
403        socket: fuchsia_async::Socket,
404    ) {
405        if let Some((buffer, _)) = self.closing_pending.remove(&(channel_id, generation)) {
406            self.connect_socket(channel_id, generation, socket, buffer, true);
407            return;
408        }
409        let is_matching_pending = matches!(
410            self.active.get(&channel_id),
411            Some(ChannelState::Pending { generation: active_gen, .. }) if *active_gen == generation
412        );
413        if !is_matching_pending {
414            return;
415        }
416        if let Some(ChannelState::Pending { buffer, .. }) = self.active.remove(&channel_id) {
417            self.connect_socket(channel_id, generation, socket, buffer, false);
418        }
419    }
420
421    /// Dispatches an [`InternalEvent`] emitted by asynchronous background workers.
422    fn handle_internal_event(&mut self, event: InternalEvent) {
423        match event {
424            InternalEvent::Registered { channel_id, generation, socket } => {
425                self.handle_registered_event(channel_id, generation, socket);
426            }
427            InternalEvent::RegistrationFailed { channel_id, generation, error } => {
428                if self.closing_pending.remove(&(channel_id, generation)).is_some() {
429                    return;
430                }
431                let matches_gen = matches!(
432                    self.active.get(&channel_id),
433                    Some(ChannelState::Pending { generation: active_gen, .. }) if *active_gen == generation
434                );
435                if matches_gen && self.active.remove(&channel_id).is_some() {
436                    let reason = format!(
437                        "Failed to register RCS socket for channel {channel_id}: {error:?}"
438                    );
439                    error!("{reason}");
440                    self.enqueue_close(channel_id, Some(reason));
441                }
442            }
443            InternalEvent::WriterDone { channel_id, generation, error } => {
444                self.closing_writers.remove(&(channel_id, generation));
445                let matches_gen = matches!(
446                    self.active.get(&channel_id),
447                    Some(ChannelState::Connected { generation: active_gen, .. }) if *active_gen == generation
448                );
449                if matches_gen && self.active.remove(&channel_id).is_some() {
450                    let reason = error.map(|e| {
451                        let msg = format!("RCS socket write error for channel {channel_id}: {e:?}");
452                        error!("{msg}");
453                        msg
454                    });
455                    self.enqueue_close(channel_id, reason);
456                }
457            }
458        }
459    }
460
461    /// Initiates opening a new channel upon receiving the first serial data chunk.
462    ///
463    /// Allocates a new generation, enforces staging byte limits and channel count limits,
464    /// stages `data` in a pending buffer, and spawns the asynchronous socket registration task.
465    fn open_pending_channel(
466        &mut self,
467        channel_id: u16,
468        data: Vec<u8>,
469        rcs_connector: &ConnectorProxy,
470    ) {
471        if self.active.len() >= MAX_ACTIVE_CHANNELS {
472            let reason = format!(
473                "Max active channels ({MAX_ACTIVE_CHANNELS}) reached; rejecting new channel {channel_id}"
474            );
475            warn!("{reason}");
476            self.enqueue_close(channel_id, Some(reason));
477            return;
478        }
479        if data.len() > MAX_CHANNEL_STAGING_BYTES {
480            let reason = format!(
481                "Initial frame on channel {channel_id} exceeds max staging bytes ({MAX_CHANNEL_STAGING_BYTES})"
482            );
483            error!("{reason}");
484            self.enqueue_close(channel_id, Some(reason));
485            return;
486        }
487        let generation = self.alloc_generation();
488        let pending_bytes = data.len();
489        let _registration_task = spawn_channel_registration(
490            channel_id,
491            generation,
492            rcs_connector.clone(),
493            self.internal_tx.clone(),
494        );
495        self.active.insert(
496            channel_id,
497            ChannelState::Pending {
498                generation,
499                buffer: vec![data],
500                pending_bytes,
501                _registration_task,
502            },
503        );
504    }
505
506    /// Processes incoming [`SerialEvent::SerialData`] for a channel.
507    ///
508    /// Directs data to the connected writer sink (staging if temporarily blocked), appends
509    /// to pending buffers if registration is in flight, or opens a new pending channel.
510    fn handle_serial_data_event(
511        &mut self,
512        channel_id: u16,
513        data: Vec<u8>,
514        rcs_connector: &ConnectorProxy,
515    ) {
516        if channel_id == CONTROL_CHANNEL_ID {
517            warn!("Ignoring SerialData for CONTROL_CHANNEL_ID in coordinator");
518            return;
519        }
520        self.drain_pending_incoming();
521        match self.active.get_mut(&channel_id) {
522            Some(ChannelState::Connected { sender, pending_incoming, pending_bytes, .. }) => {
523                if let Err(reason) = stage_or_send_connected(
524                    channel_id,
525                    data,
526                    sender,
527                    pending_incoming,
528                    pending_bytes,
529                ) {
530                    self.active.remove(&channel_id);
531                    self.enqueue_close(channel_id, Some(reason));
532                }
533            }
534            Some(ChannelState::Pending { buffer, pending_bytes, .. }) => {
535                if let Err(reason) = stage_pending_data(channel_id, data, buffer, pending_bytes) {
536                    self.active.remove(&channel_id);
537                    self.enqueue_close(channel_id, Some(reason));
538                }
539            }
540            None => self.open_pending_channel(channel_id, data, rcs_connector),
541        }
542    }
543
544    /// Transitions a connected writer into the `closing_writers` table to flush any queued
545    /// incoming data before terminating.
546    fn drain_connected_writer(
547        &mut self,
548        channel_id: u16,
549        generation: u64,
550        mut sender: mpsc::Sender<Vec<u8>>,
551        pending_incoming: VecDeque<Vec<u8>>,
552        writer_task: fuchsia_async::Task<()>,
553    ) {
554        if pending_incoming.is_empty() {
555            sender.close_channel();
556            self.closing_writers.insert((channel_id, generation), writer_task);
557        } else {
558            let drain_task = fuchsia_async::Task::spawn(async move {
559                for chunk in pending_incoming {
560                    if sender.send(chunk).await.is_err() {
561                        break;
562                    }
563                }
564                sender.close_channel();
565                writer_task.await;
566            });
567            self.closing_writers.insert((channel_id, generation), drain_task);
568        }
569    }
570
571    /// Handles a [`SerialEvent::SerialClose`] indicating the remote peer closed the channel.
572    ///
573    /// Removes the channel from active routing, discards outgoing messages, and gracefully
574    /// drains or cancels in-flight pending buffers and writer tasks.
575    fn handle_serial_close(&mut self, channel_id: u16) {
576        if channel_id == CONTROL_CHANNEL_ID {
577            warn!("Ignoring SerialClose for CONTROL_CHANNEL_ID in coordinator");
578            return;
579        }
580        self.drain_pending_incoming();
581        self.outgoing.remove(&channel_id);
582        match self.active.remove(&channel_id) {
583            Some(ChannelState::Pending { generation, buffer, _registration_task, .. })
584                if !buffer.is_empty() =>
585            {
586                self.closing_pending.insert((channel_id, generation), (buffer, _registration_task));
587            }
588            Some(ChannelState::Connected {
589                generation,
590                sender,
591                pending_incoming,
592                _writer_task,
593                ..
594            }) => {
595                self.drain_connected_writer(
596                    channel_id,
597                    generation,
598                    sender,
599                    pending_incoming,
600                    _writer_task,
601                );
602            }
603            _ => {}
604        }
605    }
606
607    /// Dispatches a [`SerialEvent`] received from the deserializer stream.
608    fn handle_serial_event(&mut self, event: SerialEvent, rcs_connector: &ConnectorProxy) {
609        match event {
610            SerialEvent::SerialData { channel_id, data } => {
611                self.handle_serial_data_event(channel_id, data, rcs_connector);
612            }
613            SerialEvent::SerialClose { channel_id } => self.handle_serial_close(channel_id),
614        }
615    }
616
617    /// Processes a [`ClientEvent`] emitted by a channel's RCS socket reader task.
618    ///
619    /// Enqueues egress data into the channel's outgoing serial queue or handles socket EOF / closure.
620    fn handle_client_event(&mut self, event: ClientEvent) {
621        let (channel_id, generation) = match &event {
622            ClientEvent::Data { channel_id, generation, .. }
623            | ClientEvent::Closed { channel_id, generation, .. } => (*channel_id, *generation),
624        };
625        let is_current = matches!(
626            self.active.get(&channel_id),
627            Some(ChannelState::Connected { generation: active_gen, .. })
628                if *active_gen == generation
629        );
630        if !is_current {
631            return;
632        }
633        match event {
634            ClientEvent::Data { channel_id, payload, .. } => {
635                self.outgoing
636                    .entry(channel_id)
637                    .or_default()
638                    .push_back(SenderMessage::Data { channel_id, payload });
639            }
640            ClientEvent::Closed { channel_id, error, .. } => {
641                let reason = error.map(|e| {
642                    let msg = format!("RCS socket read error for channel {channel_id}: {e:?}");
643                    error!("{msg}");
644                    msg
645                });
646                if let Some(state) = self.active.remove(&channel_id) {
647                    if let ChannelState::Connected {
648                        generation,
649                        sender,
650                        pending_incoming,
651                        _writer_task,
652                        ..
653                    } = state
654                    {
655                        self.drain_connected_writer(
656                            channel_id,
657                            generation,
658                            sender,
659                            pending_incoming,
660                            _writer_task,
661                        );
662                    }
663                    self.enqueue_close(channel_id, reason);
664                }
665            }
666        }
667    }
668
669    /// Selects the next outgoing [`SenderMessage`] to transmit over serial using round-robin
670    /// arbitration across all channels with non-empty egress queues.
671    fn select_next_message(&mut self) -> Option<SenderMessage> {
672        let channel_id = self
673            .last_channel_id
674            .and_then(|last_id| last_id.checked_add(1))
675            .and_then(|start_id| self.outgoing.range(start_id..).next().map(|(&k, _)| k))
676            .or_else(|| self.outgoing.keys().next().copied())?;
677        let (message, is_empty) = {
678            let queue = self.outgoing.get_mut(&channel_id)?;
679            (queue.pop_front(), queue.is_empty())
680        };
681        if is_empty {
682            self.outgoing.remove(&channel_id);
683        }
684        if message.is_some() {
685            self.last_channel_id = Some(channel_id);
686        }
687        message
688    }
689
690    /// Polls readiness of the outgoing serial sender sink and transfers the next round-robin
691    /// queued message if the sink is ready to accept it.
692    fn poll_send_next_queued(
693        &mut self,
694        sender_tx: &mut mpsc::Sender<SenderMessage>,
695        cx: &mut Context<'_>,
696    ) -> Poll<()> {
697        if self.outgoing.is_empty() {
698            return Poll::Pending;
699        }
700        match futures::sink::Sink::poll_ready(Pin::new(sender_tx), cx) {
701            Poll::Ready(Ok(())) => match self.select_next_message() {
702                Some(message) => {
703                    let _ = futures::sink::Sink::start_send(Pin::new(sender_tx), message);
704                    Poll::Ready(())
705                }
706                None => Poll::Pending,
707            },
708            Poll::Ready(Err(_)) => {
709                self.outgoing.clear();
710                Poll::Ready(())
711            }
712            Poll::Pending => Poll::Pending,
713        }
714    }
715
716    /// Polls internal readiness across both incoming socket staging queues and outgoing serial
717    /// transmission queues.
718    fn poll_ready_queues(
719        &mut self,
720        sender_tx: &mut mpsc::Sender<SenderMessage>,
721        cx: &mut Context<'_>,
722    ) -> Poll<()> {
723        if self.poll_pending_incoming(cx).is_ready() {
724            return Poll::Ready(());
725        }
726        self.poll_send_next_queued(sender_tx, cx)
727    }
728}
729
730/// Awaits the next [`ClientEvent`] from active RCS socket readers.
731///
732/// Applies backpressure by stalling if the total number of queued outgoing messages across
733/// all channels exceeds [`DEFAULT_CLIENT_DATA_CAPACITY`].
734async fn next_client_event(
735    client_rx: &mut mpsc::Receiver<ClientEvent>,
736    total_queued: usize,
737) -> Option<ClientEvent> {
738    if total_queued < DEFAULT_CLIENT_DATA_CAPACITY {
739        client_rx.next().await
740    } else {
741        futures::future::pending().await
742    }
743}
744
745/// Multiplexes FDomain channels between incoming [`SerialEvent`]s and RCS toolbox
746/// sockets via `rcs_connector`, scheduling outgoing [`SenderMessage`]s across
747/// active channels in round-robin order.
748///
749/// # Errors
750///
751/// Returns `Ok(())` when `serial_rx` closes cleanly.
752pub async fn coordinator_task(
753    mut serial_rx: mpsc::Receiver<SerialEvent>,
754    mut sender_tx: mpsc::Sender<SenderMessage>,
755    rcs_connector: ConnectorProxy,
756) -> Result<()> {
757    let (internal_tx, mut internal_rx) = mpsc::channel::<InternalEvent>(DEFAULT_CHANNEL_CAPACITY);
758    let (client_tx, mut client_rx) = mpsc::channel::<ClientEvent>(DEFAULT_CLIENT_DATA_CAPACITY);
759    let mut state = CoordinatorState::new(client_tx, internal_tx);
760    loop {
761        let total_queued: usize = state.outgoing.values().map(|queue| queue.len()).sum();
762        futures::select_biased! {
763            internal_event = internal_rx.next() => if let Some(event) = internal_event {
764                state.handle_internal_event(event);
765            },
766            _ = futures::future::poll_fn(|cx| state.poll_ready_queues(&mut sender_tx, cx)).fuse() => {}
767            client_event = next_client_event(&mut client_rx, total_queued).fuse() => if let Some(event) = client_event {
768                state.handle_client_event(event);
769            },
770            serial_event = serial_rx.next() => match serial_event {
771                Some(event) => state.handle_serial_event(event, &rcs_connector),
772                None => return Ok(()),
773            },
774        }
775    }
776}
777
778#[cfg(test)]
779mod tests {
780    use super::*;
781    use fidl_fuchsia_developer_remotecontrol_connector::{ConnectorMarker, ConnectorRequest};
782
783    #[fuchsia::test]
784    async fn test_coordinator_cancels_tasks_on_channel_close() {
785        let (mut serial_tx, serial_rx) = mpsc::channel(10);
786        let (sender_tx, mut sender_rx) = mpsc::channel(10);
787        let (connector_proxy, mut connector_stream) =
788            fidl::endpoints::create_proxy_and_stream::<ConnectorMarker>();
789        let coordinator_handle = fuchsia_async::Task::spawn(async move {
790            coordinator_task(serial_rx, sender_tx, connector_proxy).await
791        });
792        let channel_id = 42u16;
793        serial_tx
794            .send(SerialEvent::SerialData { channel_id, data: b"init".to_vec() })
795            .await
796            .unwrap();
797        let Some(Ok(ConnectorRequest::FdomainToolboxSocket { socket, responder })) =
798            connector_stream.next().await
799        else {
800            panic!("Expected FdomainToolboxSocket request");
801        };
802        responder.send().unwrap();
803        let mut rcs_socket = fuchsia_async::Socket::from_socket(socket);
804        let mut init_buf = [0u8; 4];
805        rcs_socket.read_exact(&mut init_buf).await.unwrap();
806        assert_eq!(&init_buf, b"init");
807
808        rcs_socket.write_all(b"hello").await.unwrap();
809        assert_eq!(
810            sender_rx.next().await.unwrap(),
811            SenderMessage::Data { channel_id, payload: b"hello".to_vec() }
812        );
813        drop(rcs_socket);
814        assert_eq!(
815            sender_rx.next().await.unwrap(),
816            SenderMessage::Close { channel_id, reason: None }
817        );
818        drop(serial_tx);
819        coordinator_handle.await.unwrap();
820    }
821
822    #[fuchsia::test]
823    async fn test_coordinator_flushes_buffered_data_on_serial_close() {
824        let (mut serial_tx, serial_rx) = mpsc::channel(10);
825        let (sender_tx, _sender_rx) = mpsc::channel(10);
826        let (connector_proxy, mut connector_stream) =
827            fidl::endpoints::create_proxy_and_stream::<ConnectorMarker>();
828        let coordinator_handle = fuchsia_async::Task::spawn(async move {
829            coordinator_task(serial_rx, sender_tx, connector_proxy).await
830        });
831        let channel_id = 7u16;
832        serial_tx
833            .send(SerialEvent::SerialData { channel_id, data: b"flush_me".to_vec() })
834            .await
835            .unwrap();
836        serial_tx.send(SerialEvent::SerialClose { channel_id }).await.unwrap();
837
838        let Some(Ok(ConnectorRequest::FdomainToolboxSocket { socket, responder })) =
839            connector_stream.next().await
840        else {
841            panic!("Expected FdomainToolboxSocket request");
842        };
843        responder.send().unwrap();
844        let mut rcs_socket = fuchsia_async::Socket::from_socket(socket);
845        let mut received = Vec::new();
846        rcs_socket.read_to_end(&mut received).await.unwrap();
847        assert_eq!(received, b"flush_me");
848        drop(serial_tx);
849        coordinator_handle.await.unwrap();
850    }
851
852    #[fuchsia::test]
853    async fn test_coordinator_channel_id_reuse_preserves_closing_generation() {
854        let (mut serial_tx, serial_rx) = mpsc::channel(10);
855        let (sender_tx, _sender_rx) = mpsc::channel(10);
856        let (connector_proxy, mut connector_stream) =
857            fidl::endpoints::create_proxy_and_stream::<ConnectorMarker>();
858        let coordinator_handle = fuchsia_async::Task::spawn(async move {
859            coordinator_task(serial_rx, sender_tx, connector_proxy).await
860        });
861        let channel_id = 11u16;
862        // Generation 1 opens, buffers data, and closes before registration completes.
863        serial_tx
864            .send(SerialEvent::SerialData { channel_id, data: b"gen1_payload".to_vec() })
865            .await
866            .unwrap();
867        serial_tx.send(SerialEvent::SerialClose { channel_id }).await.unwrap();
868        // Immediately reuse the same `channel_id` for Generation 2.
869        serial_tx
870            .send(SerialEvent::SerialData { channel_id, data: b"gen2_payload".to_vec() })
871            .await
872            .unwrap();
873
874        // Both registration requests should arrive and receive their respective payloads.
875        let Some(Ok(ConnectorRequest::FdomainToolboxSocket { socket: sock1, responder: resp1 })) =
876            connector_stream.next().await
877        else {
878            panic!("Expected first FdomainToolboxSocket request");
879        };
880        let Some(Ok(ConnectorRequest::FdomainToolboxSocket { socket: sock2, responder: resp2 })) =
881            connector_stream.next().await
882        else {
883            panic!("Expected second FdomainToolboxSocket request");
884        };
885        resp1.send().unwrap();
886        resp2.send().unwrap();
887
888        let mut rcs_sock1 = fuchsia_async::Socket::from_socket(sock1);
889        let mut gen1_received = Vec::new();
890        rcs_sock1.read_to_end(&mut gen1_received).await.unwrap();
891        assert_eq!(gen1_received, b"gen1_payload");
892
893        let mut rcs_sock2 = fuchsia_async::Socket::from_socket(sock2);
894        let mut gen2_buf = [0u8; 12];
895        rcs_sock2.read_exact(&mut gen2_buf).await.unwrap();
896        assert_eq!(&gen2_buf, b"gen2_payload");
897
898        drop(serial_tx);
899        coordinator_handle.await.unwrap();
900    }
901
902    #[fuchsia::test]
903    async fn test_coordinator_enforces_pending_staging_limit() {
904        let (mut serial_tx, serial_rx) = mpsc::channel(10);
905        let (sender_tx, mut sender_rx) = mpsc::channel(10);
906        let (connector_proxy, mut connector_stream) =
907            fidl::endpoints::create_proxy_and_stream::<ConnectorMarker>();
908        let coordinator_handle = fuchsia_async::Task::spawn(async move {
909            coordinator_task(serial_rx, sender_tx, connector_proxy).await
910        });
911        let channel_id = 9u16;
912        let large_chunk = vec![0xAAu8; (MAX_CHANNEL_STAGING_BYTES / 2) + 1];
913        serial_tx
914            .send(SerialEvent::SerialData { channel_id, data: large_chunk.clone() })
915            .await
916            .unwrap();
917        serial_tx.send(SerialEvent::SerialData { channel_id, data: large_chunk }).await.unwrap();
918        assert_eq!(
919            sender_rx.next().await.unwrap(),
920            SenderMessage::Close {
921                channel_id,
922                reason: Some(format!(
923                    "Pending channel {channel_id} exceeded max staging buffer ({MAX_CHANNEL_STAGING_BYTES} bytes)"
924                )),
925            }
926        );
927        assert!(connector_stream.next().now_or_never().is_none());
928
929        drop(serial_tx);
930        coordinator_handle.await.unwrap();
931    }
932
933    #[fuchsia::test]
934    async fn test_coordinator_propagates_registration_failure_reason() {
935        let (mut serial_tx, serial_rx) = mpsc::channel(10);
936        let (sender_tx, mut sender_rx) = mpsc::channel(10);
937        let (connector_proxy, connector_stream) =
938            fidl::endpoints::create_proxy_and_stream::<ConnectorMarker>();
939        drop(connector_stream);
940        let coordinator_handle = fuchsia_async::Task::spawn(async move {
941            coordinator_task(serial_rx, sender_tx, connector_proxy).await
942        });
943        let channel_id = 12u16;
944        serial_tx
945            .send(SerialEvent::SerialData { channel_id, data: b"hello".to_vec() })
946            .await
947            .unwrap();
948        match sender_rx.next().await.unwrap() {
949            SenderMessage::Close { channel_id: id, reason: Some(reason) } => {
950                assert_eq!(id, channel_id);
951                assert!(
952                    reason.contains("Failed to register RCS socket for channel 12"),
953                    "unexpected reason: {reason}"
954                );
955            }
956            other => panic!("Expected Close with reason, got {other:?}"),
957        }
958        drop(serial_tx);
959        coordinator_handle.await.unwrap();
960    }
961
962    #[fuchsia::test]
963    async fn test_select_next_message_round_robin_when_channel_drained() {
964        let (client_tx, _client_rx) = mpsc::channel(10);
965        let (internal_tx, _internal_rx) = mpsc::channel(10);
966        let mut state = CoordinatorState::new(client_tx, internal_tx);
967
968        state
969            .outgoing
970            .entry(1)
971            .or_default()
972            .push_back(SenderMessage::Data { channel_id: 1, payload: b"ch1_a".to_vec() });
973        state
974            .outgoing
975            .entry(1)
976            .or_default()
977            .push_back(SenderMessage::Data { channel_id: 1, payload: b"ch1_b".to_vec() });
978        state
979            .outgoing
980            .entry(5)
981            .or_default()
982            .push_back(SenderMessage::Data { channel_id: 5, payload: b"ch5_only".to_vec() });
983        state
984            .outgoing
985            .entry(10)
986            .or_default()
987            .push_back(SenderMessage::Data { channel_id: 10, payload: b"ch10_a".to_vec() });
988        state
989            .outgoing
990            .entry(10)
991            .or_default()
992            .push_back(SenderMessage::Data { channel_id: 10, payload: b"ch10_b".to_vec() });
993
994        assert_eq!(
995            state.select_next_message(),
996            Some(SenderMessage::Data { channel_id: 1, payload: b"ch1_a".to_vec() })
997        );
998        // Draining channel 5 removes key 5 from `self.outgoing`.
999        assert_eq!(
1000            state.select_next_message(),
1001            Some(SenderMessage::Data { channel_id: 5, payload: b"ch5_only".to_vec() })
1002        );
1003        assert!(!state.outgoing.contains_key(&5));
1004        // Next message must advance to channel 10 rather than resetting to channel 1.
1005        assert_eq!(
1006            state.select_next_message(),
1007            Some(SenderMessage::Data { channel_id: 10, payload: b"ch10_a".to_vec() })
1008        );
1009        assert_eq!(
1010            state.select_next_message(),
1011            Some(SenderMessage::Data { channel_id: 1, payload: b"ch1_b".to_vec() })
1012        );
1013        assert_eq!(
1014            state.select_next_message(),
1015            Some(SenderMessage::Data { channel_id: 10, payload: b"ch10_b".to_vec() })
1016        );
1017        assert_eq!(state.select_next_message(), None);
1018    }
1019
1020    #[fuchsia::test]
1021    async fn test_client_close_drains_connected_writer_gracefully() {
1022        let (client_tx, _client_rx) = mpsc::channel(10);
1023        let (internal_tx, mut internal_rx) = mpsc::channel(10);
1024        let mut state = CoordinatorState::new(client_tx, internal_tx);
1025
1026        let channel_id = 15u16;
1027        let generation = 1u64;
1028        let (local_socket, remote_socket) = fidl::Socket::create_stream();
1029        let socket = fuchsia_async::Socket::from_socket(local_socket);
1030        state.connect_socket(channel_id, generation, socket, vec![b"initial_".to_vec()], false);
1031
1032        if let Some(ChannelState::Connected { pending_incoming, pending_bytes, .. }) =
1033            state.active.get_mut(&channel_id)
1034        {
1035            pending_incoming.push_back(b"staged".to_vec());
1036            *pending_bytes = 6;
1037        } else {
1038            panic!("Expected channel to be in Connected state");
1039        }
1040
1041        state.handle_client_event(ClientEvent::Closed { channel_id, generation, error: None });
1042        assert!(!state.active.contains_key(&channel_id));
1043        assert!(state.closing_writers.contains_key(&(channel_id, generation)));
1044        assert_eq!(
1045            state.select_next_message(),
1046            Some(SenderMessage::Close { channel_id, reason: None })
1047        );
1048
1049        let mut rcs_socket = fuchsia_async::Socket::from_socket(remote_socket);
1050        let mut received = [0u8; 14];
1051        rcs_socket.read_exact(&mut received).await.unwrap();
1052        assert_eq!(&received, b"initial_staged");
1053
1054        match internal_rx.next().await {
1055            Some(InternalEvent::WriterDone { channel_id: id, generation: done_gen, error }) => {
1056                assert_eq!(id, channel_id);
1057                assert_eq!(done_gen, generation);
1058                assert!(error.is_none());
1059                state.handle_internal_event(InternalEvent::WriterDone {
1060                    channel_id: id,
1061                    generation: done_gen,
1062                    error,
1063                });
1064            }
1065            other => panic!("Expected WriterDone, got {:?}", other),
1066        }
1067        assert!(!state.closing_writers.contains_key(&(channel_id, generation)));
1068    }
1069}
fdomain_uart_driver_lib/coordinator.rs

fdomain_uart_driver_lib/
coordinator.rs
