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

starnix_core/vfs/socket/
socket_qipcrtr.rs
