starnix_core/vfs/socket/
socket_qipcrtr.rs

1// Copyright 2025 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::task::{
6    CurrentTask, EventHandler, SignalHandler, SignalHandlerInner, WaitCanceler, Waiter,
7};
8use crate::vfs::buffers::{AncillaryData, InputBuffer, MessageReadInfo, OutputBuffer};
9use crate::vfs::socket::{
10    SockOptValue, Socket, SocketAddress, SocketHandle, SocketMessageFlags, SocketOps, SocketPeer,
11    SocketShutdownFlags, SocketType,
12};
13use anyhow::Context;
14use fidl::endpoints::{SynchronousProxy, create_sync_proxy};
15use fidl_fuchsia_hardware_qualcomm_router as fqrtr;
16use starnix_logging::{log_warn, track_stub};
17use starnix_sync::{LockDepGuard, LockDepMutex, MappedLockDepGuard, QipcrtrSocketInnerLock};
18use starnix_uapi::errors::{Errno, from_status_like_fdio};
19use starnix_uapi::vfs::FdEvents;
20use starnix_uapi::{
21    AF_QIPCRTR, SO_RCVBUF, SO_SNDBUF, SOL_SOCKET, errno, error, sockaddr_qrtr, socklen_t, ucred,
22};
23use zerocopy::{FromBytes, IntoBytes};
24
25const QRTR_CLIENT_SERVICE_DIRECTORY: &str = "/svc/fuchsia.hardware.qualcomm.router.ClientService";
26fn connect_to_connector() -> Result<fqrtr::QrtrConnectorSynchronousProxy, anyhow::Error> {
27    let mut dir = std::fs::read_dir(QRTR_CLIENT_SERVICE_DIRECTORY)
28        .context("Failed to read ClientService directory")?;
29    let entry = dir
30        .next()
31        .ok_or_else(|| anyhow::format_err!("Missing ClientService instance"))?
32        .context("Unable to read ClientService instance")?;
33    let path = entry
34        .path()
35        .join("qrtr_connector")
36        .into_os_string()
37        .into_string()
38        .map_err(|_| anyhow::format_err!("Failed to get qrtr_connector path"))?;
39
40    let (client_end, server_end) = zx::Channel::create();
41    fdio::service_connect(&path, server_end)?;
42    Ok(fqrtr::QrtrConnectorSynchronousProxy::from_channel(client_end))
43}
44
45// From socket(7).
46pub const SEND_BUF_MIN_SIZE: usize = 2048;
47pub const SEND_BUF_MAX_SIZE: usize = 1 << 31;
48pub const SEND_BUF_DEFAULT_SIZE: usize = 2048;
49
50// From socket(7).
51pub const RECV_BUF_MIN_SIZE: usize = 256;
52pub const RECV_BUF_MAX_SIZE: usize = 1 << 31;
53pub const RECV_BUF_DEFAULT_SIZE: usize = 256;
54
55pub struct QipcrtrSocket {
56    inner: LockDepMutex<Option<QipcrtrSocketInner>, QipcrtrSocketInnerLock>,
57}
58
59struct QipcrtrSocketInner {
60    /// The proxy representing the socket in the QRTR driver.
61    proxy: fqrtr::QrtrClientConnectionSynchronousProxy,
62
63    /// The event pair representing the readable and writable signals.
64    events: zx::EventPair,
65
66    /// The peer for a connected socket, which is the default address to send messages to when no
67    /// destination is given.
68    peer: Option<sockaddr_qrtr>,
69
70    /// The socket's send buffer size.
71    ///
72    /// This value is only used to serve getsockopt calls for `SO_SNDBUF`. It does not yet enforce
73    /// a limit on the buffer size.
74    /// TODO(https://fxbug.dev/478337980): Limit the size of the send buffer.
75    send_buf_size: usize,
76
77    /// The socket's receive buffer size.
78    ///
79    /// This value is only used to serve getsockopt calls for `SO_RCVBUF`. It does not yet enforce
80    /// a limit on the buffer size.
81    /// TODO(https://fxbug.dev/478337980): Limit the size of the receive buffer.
82    recv_buf_size: usize,
83}
84
85impl QipcrtrSocket {
86    pub fn new(_socket_type: SocketType) -> Self {
87        Self { inner: Default::default() }
88    }
89
90    /// Locks and returns the inner state of the socket. If the socket is not connected to the
91    /// driver, a connection will be established, binding to an ephemeral port number.
92    fn connecting_lock(&self) -> Result<MappedLockDepGuard<'_, QipcrtrSocketInner>, Errno> {
93        let mut inner = self.inner.lock();
94        if inner.is_none() {
95            *inner = Some(QipcrtrSocketInner::new(fqrtr::ConnectionOptions {
96                blocking: Some(false),
97                ..Default::default()
98            })?);
99        }
100        Ok(LockDepGuard::map(inner, |inner| inner.as_mut().unwrap()))
101    }
102
103    fn close(&self) {
104        *self.inner.lock() = None;
105    }
106}
107
108impl QipcrtrSocketInner {
109    fn new(options: fqrtr::ConnectionOptions) -> Result<Self, Errno> {
110        let connector = connect_to_connector().map_err(|e| errno!(ENETUNREACH, e))?;
111
112        let (client_end, server_end) = create_sync_proxy::<fqrtr::QrtrClientConnectionMarker>();
113        connector
114            .get_connection(&options, server_end, zx::MonotonicInstant::INFINITE)
115            .map_err(|e| errno!(ENETUNREACH, e))?
116            .map_err(qrtr_error_to_errno)?;
117
118        let proxy = fqrtr::QrtrClientConnectionSynchronousProxy::new(client_end.into_channel());
119        let events = proxy
120            .get_signals(zx::MonotonicInstant::INFINITE)
121            .map_err(|e| errno!(ENETUNREACH, e))?;
122
123        Ok(Self {
124            proxy,
125            events,
126            peer: None,
127            send_buf_size: SEND_BUF_DEFAULT_SIZE,
128            recv_buf_size: RECV_BUF_DEFAULT_SIZE,
129        })
130    }
131
132    /// Returns the [`sockaddr_qrtr`] of this connection.
133    fn bound_addr(&self) -> Result<sockaddr_qrtr, Errno> {
134        let addr = sockaddr_qrtr {
135            sq_family: AF_QIPCRTR,
136            sq_node: self
137                .proxy
138                .get_node_id(zx::MonotonicInstant::INFINITE)
139                .map_err(|e| errno!(EINVAL, e))?,
140            sq_port: self
141                .proxy
142                .get_port_id(zx::MonotonicInstant::INFINITE)
143                .map_err(|e| errno!(EINVAL, e))?,
144            ..Default::default()
145        };
146        Ok(addr)
147    }
148}
149
150impl Drop for QipcrtrSocketInner {
151    fn drop(&mut self) {
152        if let Err(e) = self.proxy.close_connection(zx::MonotonicInstant::INFINITE) {
153            log_warn!("Failed to close QRTR connection: {e:?}");
154        }
155    }
156}
157
158impl SocketOps for QipcrtrSocket {
159    fn connect(
160        &self,
161        _socket: &SocketHandle,
162        _current_task: &CurrentTask,
163        peer: SocketPeer,
164    ) -> Result<(), Errno> {
165        let peer = match peer {
166            SocketPeer::Address(addr) => extract_qrtr_sockaddr(&addr)?,
167            _ => {
168                return error!(EINVAL);
169            }
170        };
171
172        let mut inner = self.inner.lock();
173        if inner.is_some() {
174            return error!(EISCONN);
175        }
176
177        // Establish a connection without a specific port number. The driver will automatically
178        // assign one, resulting in a bound socket.
179        let mut new_inner = QipcrtrSocketInner::new(fqrtr::ConnectionOptions {
180            blocking: Some(false),
181            ..Default::default()
182        })?;
183        new_inner.peer = Some(peer);
184
185        *inner = Some(new_inner);
186        Ok(())
187    }
188
189    fn listen(&self, _socket: &Socket, _backlog: i32, _credentials: ucred) -> Result<(), Errno> {
190        error!(ENOTSUP)
191    }
192
193    fn accept(&self, _socket: &Socket, _current_task: &CurrentTask) -> Result<SocketHandle, Errno> {
194        error!(ENOTSUP)
195    }
196
197    fn bind(
198        &self,
199        _socket: &Socket,
200        _current_task: &CurrentTask,
201        socket_address: SocketAddress,
202    ) -> Result<(), Errno> {
203        let addr = extract_qrtr_sockaddr(&socket_address)?;
204
205        let mut inner = self.inner.lock();
206        if inner.is_some() {
207            return error!(EINVAL);
208        }
209
210        // Establish a connection with the specified port number.
211        *inner = Some(QipcrtrSocketInner::new(fqrtr::ConnectionOptions {
212            blocking: Some(false),
213            port: Some(addr.sq_port),
214            ..Default::default()
215        })?);
216
217        Ok(())
218    }
219
220    fn read(
221        &self,
222        _socket: &Socket,
223        _current_task: &CurrentTask,
224        data: &mut dyn OutputBuffer,
225        flags: SocketMessageFlags,
226    ) -> Result<MessageReadInfo, Errno> {
227        if flags.contains(SocketMessageFlags::PEEK) {
228            track_stub!(
229                TODO("https://fxbug.dev/388082019"),
230                "SocketMessageFlags::PEEK is unsupported"
231            );
232            return error!(EINVAL);
233        }
234
235        let inner = self.connecting_lock()?;
236
237        if flags.contains(SocketMessageFlags::DONTWAIT) {
238            match inner.events.wait_one(
239                zx::Signals::from_bits_truncate(fqrtr::SIGNAL_READABLE)
240                    | zx::Signals::EVENTPAIR_PEER_CLOSED,
241                zx::MonotonicInstant::INFINITE_PAST,
242            ) {
243                zx::WaitResult::Ok(_) => {}
244                zx::WaitResult::TimedOut(_) | zx::WaitResult::Canceled(_) => return error!(EAGAIN),
245                zx::WaitResult::Err(status) => return Err(from_status_like_fdio!(status)),
246            }
247        }
248
249        let (src_node, src_port, src_data) = inner
250            .proxy
251            .read(zx::MonotonicInstant::INFINITE)
252            .map_err(|e| errno!(ECONNRESET, e))?
253            .map_err(qrtr_error_to_errno)?;
254
255        let bytes_read = data.write(src_data.as_bytes())?;
256        Ok(MessageReadInfo {
257            bytes_read,
258            message_length: src_data.len(),
259            address: Some(pack_qrtr_sockaddr(src_node, src_port)),
260            ..Default::default()
261        })
262    }
263
264    fn write(
265        &self,
266        _socket: &Socket,
267        _current_task: &CurrentTask,
268        data: &mut dyn InputBuffer,
269        dest_address: &mut Option<SocketAddress>,
270        _ancillary_data: &mut Vec<AncillaryData>,
271    ) -> Result<usize, Errno> {
272        let inner = self.connecting_lock()?;
273
274        // If no destination address is specified, send to the peer address, which is set if
275        // connect() is called.
276        let dest = match dest_address {
277            Some(addr) => extract_qrtr_sockaddr(addr)?,
278            None => inner.peer.ok_or_else(|| errno!(EDESTADDRREQ))?,
279        };
280
281        match inner.events.wait_one(
282            zx::Signals::from_bits_truncate(fqrtr::SIGNAL_WRITABLE)
283                | zx::Signals::EVENTPAIR_PEER_CLOSED,
284            zx::MonotonicInstant::INFINITE_PAST,
285        ) {
286            zx::WaitResult::Ok(_) => {}
287            zx::WaitResult::TimedOut(_) | zx::WaitResult::Canceled(_) => return error!(EAGAIN),
288            zx::WaitResult::Err(status) => return Err(from_status_like_fdio!(status)),
289        }
290
291        let data_written = data.read_all()?;
292        let _ = inner
293            .proxy
294            .write(
295                dest.sq_node,
296                dest.sq_port,
297                data_written.as_ref(),
298                zx::MonotonicInstant::INFINITE,
299            )
300            .map_err(|e| errno!(ECONNRESET, e))?
301            .map_err(qrtr_error_to_errno)?;
302        Ok(data_written.len())
303    }
304
305    fn wait_async(
306        &self,
307        _socket: &Socket,
308        _current_task: &CurrentTask,
309        waiter: &Waiter,
310        events: FdEvents,
311        handler: EventHandler,
312    ) -> WaitCanceler {
313        let Ok(inner) = self.connecting_lock() else {
314            return WaitCanceler::new_noop();
315        };
316        let signal_handler = SignalHandler {
317            inner: SignalHandlerInner::ZxHandle(qrtr_signals_to_fd_events),
318            event_handler: handler,
319            err_code: None,
320        };
321        let canceler = waiter
322            .wake_on_zircon_signals(
323                &inner.events,
324                fd_events_to_qrtr_signals(events),
325                signal_handler,
326            )
327            .unwrap();
328        WaitCanceler::new_port(canceler)
329    }
330
331    fn query_events(
332        &self,
333        _socket: &Socket,
334        _current_task: &CurrentTask,
335    ) -> Result<FdEvents, Errno> {
336        let inner = self.connecting_lock()?;
337        let signals = inner
338            .events
339            .as_handle_ref()
340            .wait_one(zx::Signals::all(), zx::MonotonicInstant::INFINITE_PAST)
341            .map_err(|e| from_status_like_fdio!(e))?;
342        Ok(qrtr_signals_to_fd_events(signals))
343    }
344
345    fn shutdown(&self, _socket: &Socket, _how: SocketShutdownFlags) -> Result<(), Errno> {
346        self.close();
347        Ok(())
348    }
349
350    fn close(&self, _current_task: &CurrentTask, _socket: &Socket) {
351        self.close();
352    }
353
354    fn getsockname(&self, _socket: &Socket) -> Result<SocketAddress, Errno> {
355        let name = self.connecting_lock()?.bound_addr()?;
356        Ok(SocketAddress::Qipcrtr(name.as_bytes().to_vec()))
357    }
358
359    fn getpeername(&self, _socket: &Socket) -> Result<SocketAddress, Errno> {
360        let peer = self.connecting_lock()?.peer.ok_or_else(|| errno!(ENOTCONN))?;
361        Ok(SocketAddress::Qipcrtr(peer.as_bytes().to_vec()))
362    }
363
364    fn setsockopt(
365        &self,
366        _socket: &Socket,
367        current_task: &CurrentTask,
368        level: u32,
369        optname: u32,
370        optval: SockOptValue,
371    ) -> Result<(), Errno> {
372        let mut inner = self.connecting_lock()?;
373        match level {
374            SOL_SOCKET => match optname {
375                SO_SNDBUF => {
376                    let requested_capacity: socklen_t = optval.read(current_task)?;
377                    // SO_SNDBUF doubles the requested capacity to leave space for bookkeeping.
378                    // See https://man7.org/linux/man-pages/man7/socket.7.html
379                    let capacity = usize::try_from(requested_capacity * 2).unwrap_or(usize::MAX);
380                    // TODO(https://fxbug.dev/322907334): Clamp to `wmem_max`.
381                    let capacity = capacity.clamp(SEND_BUF_MIN_SIZE, SEND_BUF_MAX_SIZE);
382                    inner.send_buf_size = capacity;
383                }
384                SO_RCVBUF => {
385                    let requested_capacity: socklen_t = optval.read(current_task)?;
386                    // SO_RCVBUF doubles the requested capacity to leave space for bookkeeping.
387                    // See https://man7.org/linux/man-pages/man7/socket.7.html
388                    let capacity = usize::try_from(requested_capacity * 2).unwrap_or(usize::MAX);
389                    // TODO(https://fxbug.dev/322906968): Clamp to `rmem_max`.
390                    let capacity = capacity.clamp(RECV_BUF_MIN_SIZE, RECV_BUF_MAX_SIZE);
391                    inner.recv_buf_size = capacity;
392                }
393                _ => return error!(ENOSYS),
394            },
395            _ => return error!(ENOSYS),
396        }
397
398        Ok(())
399    }
400
401    fn getsockopt(
402        &self,
403        _socket: &Socket,
404        _current_task: &CurrentTask,
405        level: u32,
406        optname: u32,
407        _optlen: u32,
408    ) -> Result<Vec<u8>, Errno> {
409        let inner = self.connecting_lock()?;
410        Ok(match level {
411            SOL_SOCKET => match optname {
412                SO_SNDBUF => (inner.send_buf_size as socklen_t).to_ne_bytes().to_vec(),
413                SO_RCVBUF => (inner.recv_buf_size as socklen_t).to_ne_bytes().to_vec(),
414                _ => return error!(ENOSYS),
415            },
416            _ => vec![],
417        })
418    }
419}
420
421/// Returns the [`sockaddr_qrtr`] within a [`SocketAddress`] or `EINVAL` if the address is not a
422/// QRTR address.
423fn extract_qrtr_sockaddr(addr: &SocketAddress) -> Result<sockaddr_qrtr, Errno> {
424    match addr {
425        SocketAddress::Qipcrtr(bytes) => sockaddr_qrtr::read_from_prefix(bytes.as_bytes())
426            .map(|(addr, _)| addr)
427            .map_err(|e| errno!(EINVAL, e)),
428        _ => error!(EINVAL),
429    }
430}
431
432/// Returns the [`SocketAddress`] representing a given node and port number.
433fn pack_qrtr_sockaddr(node: u32, port: u32) -> SocketAddress {
434    let addr =
435        sockaddr_qrtr { sq_family: AF_QIPCRTR, sq_node: node, sq_port: port, ..Default::default() };
436    SocketAddress::Qipcrtr(addr.as_bytes().into())
437}
438
439/// Maps a [`fqrtr::Error`] to an [`Errno`]. This mapping is not one-to-one.
440fn qrtr_error_to_errno(e: fqrtr::Error) -> Errno {
441    match e {
442        fqrtr::Error::InternalError => errno!(EINVAL),
443        fqrtr::Error::AlreadyPending => errno!(EBUSY),
444        fqrtr::Error::RemoteNodeUnavailable => errno!(ECONNRESET),
445        fqrtr::Error::AlreadyBound => errno!(EADDRINUSE),
446        fqrtr::Error::NotSupported => errno!(ENOTSUP),
447        fqrtr::Error::WouldBlock => errno!(EAGAIN),
448        fqrtr::Error::NoResources => errno!(ENOMEM),
449        fqrtr::Error::InvalidArgs => errno!(EINVAL),
450        _ => errno!(EINVAL),
451    }
452}
453
454/// Maps [`FdEvents`] to [`zx::Signals`] for a QRTR connection.
455fn fd_events_to_qrtr_signals(events: FdEvents) -> zx::Signals {
456    let mut signals = zx::Signals::empty();
457    if events.contains(FdEvents::POLLIN) {
458        signals |= zx::Signals::from_bits_truncate(fqrtr::SIGNAL_READABLE);
459    }
460    if events.contains(FdEvents::POLLOUT) {
461        signals |= zx::Signals::from_bits_truncate(fqrtr::SIGNAL_WRITABLE);
462    }
463
464    // Always wait for the peer to be closed, which can generate POLLHUP.
465    signals |= zx::Signals::EVENTPAIR_PEER_CLOSED;
466    signals
467}
468
469/// Maps [`zx::Signals`] to [`FdEvents`] for a QRTR connection.
470fn qrtr_signals_to_fd_events(signals: zx::Signals) -> FdEvents {
471    let mut events = FdEvents::empty();
472    if signals.contains(zx::Signals::from_bits_truncate(fqrtr::SIGNAL_READABLE)) {
473        events |= FdEvents::POLLIN;
474    }
475    if signals.contains(zx::Signals::from_bits_truncate(fqrtr::SIGNAL_WRITABLE)) {
476        events |= FdEvents::POLLOUT;
477    }
478    if signals.contains(zx::Signals::EVENTPAIR_PEER_CLOSED) {
479        events |= FdEvents::POLLHUP;
480    }
481    events
482}
483
484#[cfg(test)]
485mod tests {
486    use super::*;
487    use crate::testing::spawn_kernel_and_run;
488    use crate::vfs::buffers::{VecInputBuffer, VecOutputBuffer};
489    use crate::vfs::socket::{SocketDomain, SocketProtocol, SocketType};
490    use fidl::endpoints::create_sync_proxy;
491    use futures::StreamExt;
492
493    /// Creates a `QipcrtrSocket` with a mock inner state.
494    ///
495    /// The mock state is connected to a `QrtrClientConnection` proxy, and the stream for that
496    /// proxy is returned to allow the test to drive the mock FIDL behavior.
497    fn mock_qipcrtr_socket()
498    -> (QipcrtrSocket, fidl::endpoints::ServerEnd<fqrtr::QrtrClientConnectionMarker>) {
499        let (proxy, server_end) = create_sync_proxy::<fqrtr::QrtrClientConnectionMarker>();
500
501        // We need an event pair for the socket.
502        let (events, _) = zx::EventPair::create();
503
504        let inner = QipcrtrSocketInner {
505            proxy,
506            events,
507            peer: None,
508            send_buf_size: SEND_BUF_DEFAULT_SIZE,
509            recv_buf_size: RECV_BUF_DEFAULT_SIZE,
510        };
511
512        (QipcrtrSocket { inner: Some(inner).into() }, server_end)
513    }
514
515    #[::fuchsia::test]
516    async fn test_qipcrtr_socket_new() {
517        spawn_kernel_and_run(async |current_task| {
518            let _kernel = current_task.kernel();
519            // This test just checks basic creation without panic, but for QIPCRTR it tries
520            // to connect to the global service, which might fail in test env if not mocked
521            // correctly or if we rely on real service. The existing test `test_qipcrtr_socket_new`
522            // calls `Socket::new` which calls `QipcrtrSocket::new`.
523            // `QipcrtrSocket::new` creates a None inner, so it doesn't connect yet.
524            // Connection happens on first use or explicit connect.
525            let _socket = Socket::new(
526                &current_task,
527                SocketDomain::Qipcrtr,
528                SocketType::Datagram,
529                SocketProtocol::default(),
530                /* kernel_private = */ false,
531            )
532            .expect("Failed to create socket.");
533        })
534        .await;
535    }
536
537    #[::fuchsia::test]
538    async fn test_qipcrtr_sockopt() {
539        spawn_kernel_and_run(async |current_task| {
540            let socket = mock_qipcrtr_socket();
541            let socket_obj = Socket::new_with_ops_and_info(
542                Box::new(socket.0),
543                SocketDomain::Qipcrtr,
544                SocketType::Datagram,
545                SocketProtocol::default(),
546            );
547            let _server_end = socket.1;
548
549            // Test SO_SNDBUF
550            let sndbuf = socket_obj.getsockopt(&current_task, SOL_SOCKET, SO_SNDBUF, 4).unwrap();
551            let sndbuf_val = u32::from_ne_bytes(sndbuf.as_slice().try_into().unwrap());
552            assert_eq!(sndbuf_val, SEND_BUF_DEFAULT_SIZE as u32);
553
554            let new_sndbuf: u32 = 4096;
555            socket_obj
556                .setsockopt(
557                    &current_task,
558                    SOL_SOCKET,
559                    SO_SNDBUF,
560                    SockOptValue::from(new_sndbuf.as_bytes().to_vec()),
561                )
562                .unwrap();
563
564            let sndbuf = socket_obj.getsockopt(&current_task, SOL_SOCKET, SO_SNDBUF, 4).unwrap();
565            let sndbuf_val = u32::from_ne_bytes(sndbuf.as_slice().try_into().unwrap());
566            // Setsockopt doubles the value.
567            assert_eq!(sndbuf_val, new_sndbuf * 2);
568
569            // Test SO_RCVBUF
570            let rcvbuf = socket_obj.getsockopt(&current_task, SOL_SOCKET, SO_RCVBUF, 4).unwrap();
571            let rcvbuf_val = u32::from_ne_bytes(rcvbuf.as_slice().try_into().unwrap());
572            assert_eq!(rcvbuf_val, RECV_BUF_DEFAULT_SIZE as u32);
573
574            let new_rcvbuf: u32 = 1024;
575            socket_obj
576                .setsockopt(
577                    &current_task,
578                    SOL_SOCKET,
579                    SO_RCVBUF,
580                    SockOptValue::from(new_rcvbuf.as_bytes().to_vec()),
581                )
582                .unwrap();
583
584            let rcvbuf = socket_obj.getsockopt(&current_task, SOL_SOCKET, SO_RCVBUF, 4).unwrap();
585            let rcvbuf_val = u32::from_ne_bytes(rcvbuf.as_slice().try_into().unwrap());
586            // Setsockopt doubles the value.
587            assert_eq!(rcvbuf_val, new_rcvbuf * 2);
588        })
589        .await;
590    }
591
592    #[::fuchsia::test]
593    async fn test_qipcrtr_sockname() {
594        let (socket_inner, server_end) = mock_qipcrtr_socket();
595        // Handle get_node_id and get_port_id requests
596        std::thread::spawn(move || {
597            let mut executor = fuchsia_async::LocalExecutor::default();
598            executor.run_singlethreaded(async move {
599                let mut stream = server_end.into_stream();
600                while let Some(Ok(request)) = stream.next().await {
601                    match request {
602                        fqrtr::QrtrClientConnectionRequest::GetNodeId { responder, .. } => {
603                            let _ = responder.send(123).unwrap();
604                        }
605                        fqrtr::QrtrClientConnectionRequest::GetPortId { responder, .. } => {
606                            let _ = responder.send(456).unwrap();
607                        }
608                        fqrtr::QrtrClientConnectionRequest::CloseConnection {
609                            responder, ..
610                        } => {
611                            let _ = responder.send();
612                        }
613                        _ => panic!("Unexpected request: {:?}", request),
614                    }
615                }
616            });
617        });
618
619        spawn_kernel_and_run(async |_current_task| {
620            let socket_obj = Socket::new_with_ops_and_info(
621                Box::new(socket_inner),
622                SocketDomain::Qipcrtr,
623                SocketType::Datagram,
624                SocketProtocol::default(),
625            );
626
627            let addr = socket_obj.getsockname().unwrap();
628            let qrtr_addr = extract_qrtr_sockaddr(&addr).unwrap();
629            assert_eq!(qrtr_addr.sq_node, 123);
630            assert_eq!(qrtr_addr.sq_port, 456);
631
632            // Set peer
633            let peer_addr = sockaddr_qrtr {
634                sq_family: AF_QIPCRTR,
635                sq_node: 10,
636                sq_port: 20,
637                ..Default::default()
638            };
639            socket_obj
640                .downcast_socket::<QipcrtrSocket>()
641                .unwrap()
642                .inner
643                .lock()
644                .as_mut()
645                .unwrap()
646                .peer = Some(peer_addr);
647
648            let peer = socket_obj.getpeername().unwrap();
649            let peer_qrtr = extract_qrtr_sockaddr(&peer).unwrap();
650            assert_eq!(peer_qrtr.sq_node, 10);
651            assert_eq!(peer_qrtr.sq_port, 20);
652        })
653        .await;
654    }
655
656    #[::fuchsia::test]
657    async fn test_qipcrtr_read_write() {
658        let (socket_inner, server_end) = mock_qipcrtr_socket();
659        std::thread::spawn(move || {
660            let mut executor = fuchsia_async::LocalExecutor::default();
661            executor.run_singlethreaded(async move {
662                let mut stream = server_end.into_stream();
663                while let Some(Ok(request)) = stream.next().await {
664                    match request {
665                        fqrtr::QrtrClientConnectionRequest::Write {
666                            dst_node_id,
667                            dst_port,
668                            data,
669                            responder,
670                            ..
671                        } => {
672                            assert_eq!(dst_node_id, 10);
673                            assert_eq!(dst_port, 20);
674                            assert_eq!(data, b"hello");
675                            let _ = responder.send(Ok(())).unwrap();
676                        }
677                        fqrtr::QrtrClientConnectionRequest::Read { responder, .. } => {
678                            let _ = responder.send(Ok((5, 15, b"world"))).unwrap();
679                        }
680                        fqrtr::QrtrClientConnectionRequest::CloseConnection {
681                            responder, ..
682                        } => {
683                            let _ = responder.send();
684                        }
685                        _ => panic!("Unexpected request: {:?}", request),
686                    }
687                }
688            });
689        });
690
691        spawn_kernel_and_run(async |current_task| {
692            let socket_obj = Socket::new_with_ops_and_info(
693                Box::new(socket_inner),
694                SocketDomain::Qipcrtr,
695                SocketType::Datagram,
696                SocketProtocol::default(),
697            );
698            // Connect to set default peer
699            let peer_addr = sockaddr_qrtr {
700                sq_family: AF_QIPCRTR,
701                sq_node: 10,
702                sq_port: 20,
703                ..Default::default()
704            };
705            socket_obj
706                .downcast_socket::<QipcrtrSocket>()
707                .unwrap()
708                .inner
709                .lock()
710                .as_mut()
711                .unwrap()
712                .peer = Some(peer_addr);
713
714            // Test Write
715            let mut input = VecInputBuffer::new(b"hello");
716            let written =
717                socket_obj.write(&current_task, &mut input, &mut None, &mut vec![]).unwrap();
718            assert_eq!(written, 5);
719
720            // Test Read
721            let mut output = VecOutputBuffer::new(100);
722            let info =
723                socket_obj.read(&current_task, &mut output, SocketMessageFlags::empty()).unwrap();
724            assert_eq!(info.bytes_read, 5);
725            assert_eq!(output.data(), b"world");
726
727            let addr = extract_qrtr_sockaddr(&info.address.unwrap()).unwrap();
728            assert_eq!(addr.sq_node, 5);
729            assert_eq!(addr.sq_port, 15);
730        })
731        .await;
732    }
733}
starnix_core/vfs/socket/socket_qipcrtr.rs

starnix_core/vfs/socket/
socket_qipcrtr.rs
