diff --git a/services/net/src/api.rs b/services/net/src/api.rs index f964d1de1..2f5904644 100644 --- a/services/net/src/api.rs +++ b/services/net/src/api.rs @@ -152,9 +152,12 @@ pub(crate) enum Opcode { /// is returned as part of the `Valid` parameter. This is not blocking. StdTcpPeek = 32, - /// Receives data from the specified TCP Connection. The TCP connection number is - /// passed in the upper 16 bits of the opcode, and the number of received bytes - /// is returned as part of the `Valid` parameter. + /// Receives data from the specified TCP Connection. + /// + /// - The TCP connection number is passed in the upper 16 bits of the opcode. + /// - The return buffer is the `buf` parameter + /// - The read timeout is passed as the `offset` parameter. + /// - The number of received bytes is returned as part of the `valid` parameter. StdTcpRx = 33, /// Close the TCP connection. The connection ID is specified in the upper 16 bits @@ -176,6 +179,81 @@ pub(crate) enum Opcode { /// BlockingScalar call to set the NODELAY / "Nagle" value of this connection StdSetNodelay = 39, + + /// Bind a UDP listener to the specified SocketAddr + /// + /// # Arguments + /// + /// u8 array + /// -------|--------- + /// offset | Contents + /// =======|========= + /// 0 | port number, low byte + /// 1 | port number, high byte + /// 2 | address type -- 4 = ipv4, 6 = ipv6 + /// ... | remaining bytes are the address + /// + /// # Returns + /// + /// u16 array + /// -------|--------- + /// offset | Contents + /// =======|========= + /// 0 | 0 indicating success + /// 1 | Connection index + /// + /// # Errors + /// + /// u8 array + /// -------|--------- + /// offset | Contents + /// =======|========= + /// 0 | 1 indicating error + /// 1 | 1 indicating error (duplicated) + /// 2 | 1 indicating error (duplicated) + /// 3 | 1 indicating error (duplicated) + /// 4 | Error code + StdUdpBind = 40, + + //StdUdpClose = 41, + + /// Receives data from the specified TCP Connection. + /// + /// - The UDP connection number is passed in the upper 16 bits of the opcode. + /// - The arguments are passed as follows: + /// u8 array + /// -------|--------- + /// offset | Contents + /// =======|========= + /// 0 | 0 indicates nonblocking; 1 indicates blocking + /// 1 | bytes:u64 LSB timeout in ms + /// ... + /// 8 | bytes:u64 LSB timeout in ms + /// - The return buffer is the `buf` parameter, which is split as follows: + /// u8 array + /// -------|--------- + /// offset | Contents + /// =======|========= + /// 0 | 0 indicates success; 1 indicates error + /// 1 | bytes:u16 received, low byte + /// 2 | bytes:u16 received, high byte + /// 3 | address type -- 4 = ipv4, 6 = ipv6, 0 = invalid/err + /// 4 | LSB ip address + /// ... + /// 7 | MSB ipv4 address + /// ... + /// 19| MSW ipv6 address + /// 20| remote port:u16, low byte + /// 21| remote port:u16, high byte + /// 22| received data byte 0 + /// ... + /// 22+BUFLEN-1| max last data byte (nominally equal to NET_MTU) + /// - The `valid` and `offset` parameters are not used. + StdUdpRx = 42, + + //StdUdpTx = 43, + //StdUdpSetTtl = 44, + //StdUdpGetTtl = 45, } #[derive(Debug, Archive, Serialize, Deserialize, Copy, Clone, Default)] @@ -250,6 +328,7 @@ pub enum NetError { LibraryError = 6, // AlreadyUsed = 7, TimedOut = 8, + WouldBlock = 9, } /////// a bunch of structures are re-derived here so we can infer `rkyv` traits on them diff --git a/services/net/src/main.rs b/services/net/src/main.rs index 9481dbd32..90204c442 100644 --- a/services/net/src/main.rs +++ b/services/net/src/main.rs @@ -7,6 +7,8 @@ mod std_tcpstream; use std_tcpstream::*; mod std_glue; use std_glue::*; +mod std_udp; +use std_udp::*; use com::api::{ComIntSources, Ipv4Conf, NET_MTU}; use num_traits::*; @@ -104,6 +106,12 @@ struct WaitingSocket { expiry: Option, } +pub struct UdpStdState { + pub msg: xous::MessageEnvelope, + pub handle: SocketHandle, + pub expiry: Option, +} + fn set_com_ints(com_int_list: &mut Vec) { com_int_list.clear(); com_int_list.push(ComIntSources::WlanIpConfigUpdate); @@ -158,11 +166,12 @@ fn xmain() -> ! { // storage for all our sockets let mut sockets = SocketSet::new(vec![]); + // ------------- libstd variant ----------- // Each process keeps track of its own sockets. These are kept in a Vec. When a handle // is destroyed, it is turned into a `None`. let mut process_sockets: HashMap, Vec>> = HashMap::new(); - // When a client issues a Receive request, it will get placed here while the packet data + // When a TCP client issues a Receive request, it will get placed here while the packet data // is being accumulated. let mut tcp_rx_waiting: Vec> = Vec::new(); @@ -182,6 +191,11 @@ fn xmain() -> ! { )>, > = Vec::new(); + // When a UDP client opens a socket, an entry is automatically created here to accumulate + // incoming UDP socket data. + let mut udp_rx_waiting: Vec> = Vec::new(); + + // ------------- native variant ----------- // ping storage // up to four concurrent pings in the queue let icmp_rx_buffer = IcmpSocketBuffer::new( @@ -986,6 +1000,27 @@ fn xmain() -> ! { } buf.replace(tcpspec).unwrap(); } + + Some(Opcode::StdUdpBind) => { + let pid = msg.sender.pid(); + std_udp_bind( + msg, + &mut sockets, + process_sockets.entry(pid).or_default(), + ); + } + + Some(Opcode::StdUdpRx) => { + let pid = msg.sender.pid(); + std_udp_rx( + msg, + &timer, + &mut sockets, + &mut udp_rx_waiting, + process_sockets.entry(pid).or_default(), + ); + } + Some(Opcode::UdpBind) => { let mut buf = unsafe { Buffer::from_memory_message_mut(msg.body.memory_message_mut().unwrap()) @@ -1646,6 +1681,58 @@ fn xmain() -> ! { } } + // this block handles StdUdp + for connection in udp_rx_waiting.iter_mut() { + let mut socket; + let UdpStdState { + mut msg, + handle: _, + expiry: _, + } = { + match connection { + &mut None => continue, + Some(s) => { + socket = sockets.get::(s.handle); + if !socket.can_recv() { + if let Some(trigger) = s.expiry { + if trigger < now { + // timer expired + } else { + continue; + } + } else { + continue; + } + } // we don't process the Rx here because we need to `take()` the message first, so that its lifetime ends + } + } + // remove the connection from the list, allowing subsequent code to operate on it and then .drop() + connection.take().unwrap() + }; + + // If it can't receive, then the only explanation was that it timed out + if !socket.can_recv() { + udp_rx_failure(msg, NetError::TimedOut); + continue; + } + + // Extract the receive data here; the `msg` will go out of scope at this point. + match socket.recv() { + Ok((data, endpoint)) => { + udp_rx_success( + // unwrap is safe here because the message was type-checked prior to insertion into the waiting queue + msg.body.memory_message_mut().unwrap().buf.as_slice_mut(), + data, + endpoint + ); + } + Err(e) => { + log::error!("unable to receive: {:?}", e); + udp_rx_failure(msg, NetError::LibraryError); + } + } + } + // this block contains the ICMP Rx handler. Tx is initiated by an incoming message to the Net crate. { let mut socket = sockets.get::(icmp_handle); diff --git a/services/net/src/std_glue.rs b/services/net/src/std_glue.rs index e7a1f9f40..41ce14d71 100644 --- a/services/net/src/std_glue.rs +++ b/services/net/src/std_glue.rs @@ -64,7 +64,9 @@ pub(crate) fn respond_with_error(mut env: xous::MessageEnvelope, code: NetError) let s: &mut [u8] = body.buf.as_slice_mut(); let mut i = s.iter_mut(); - // Duplicate error to ensure it's seen as an error regardless of byte order + // Duplicate error to ensure it's seen as an error regardless of byte order/return type + // This is necessary because errors are encoded as `u8` slices, but "good" + // responses may be encoded as `u16` or `u32` slices. *i.next()? = 1; *i.next()? = 1; *i.next()? = 1; diff --git a/services/net/src/std_udp.rs b/services/net/src/std_udp.rs new file mode 100644 index 000000000..a8f925a09 --- /dev/null +++ b/services/net/src/std_udp.rs @@ -0,0 +1,193 @@ +use crate::*; +use smoltcp::wire::{IpEndpoint, IpAddress}; +use ticktimer_server::Ticktimer; + +/// Overall architecture for libstd UDP implementation. +/// +/// Sockets are stored in the PID/SocketHandle HashMap `process_sockets` (this is shared with TCP) +/// `recv` requests create `UpdStdState` objects, that are stored in a `udp_rx` Vec. + +const BUFLEN: usize = NET_MTU as usize; + +pub(crate) fn std_udp_bind( + mut msg: xous::MessageEnvelope, + sockets: &mut SocketSet, + our_sockets: &mut Vec>, + ) { + // Ignore nonblocking and scalar messages + let body = match msg.body.memory_message_mut() { + Some(b) => b, + None => { + log::trace!("invalid message type"); + respond_with_error(msg, NetError::LibraryError); + return; + } + }; + + let bytes = body.buf.as_slice::(); + let local_port = u16::from_le_bytes([bytes[0], bytes[1]]); + let address = match parse_address(&bytes[2..]) { + Some(addr) => addr, + None => { + log::trace!("couldn't parse address"); + respond_with_error(msg, NetError::LibraryError); + return; + } + }; + + let udp_rx_buffer = + UdpSocketBuffer::new(vec![UdpPacketMetadata::EMPTY], vec![0; BUFLEN]); + let udp_tx_buffer = + UdpSocketBuffer::new(vec![UdpPacketMetadata::EMPTY], vec![0; BUFLEN]); + let mut udp_socket = UdpSocket::new(udp_rx_buffer, udp_tx_buffer); + + // Attempt to connect, returning the error if there is one + if let Err(e) = udp_socket + .bind(IpEndpoint{addr: address, port: local_port}) + .map_err(|e| match e { + smoltcp::Error::Illegal => NetError::SocketInUse, + smoltcp::Error::Unaddressable => NetError::Unaddressable, + _ => NetError::LibraryError, + }) + { + log::trace!("couldn't connect: {:?}", e); + respond_with_error(msg, e); + return; + } + + let handle = sockets.add(udp_socket); + + // Add the socket into our process' list of sockets, and pass the index back as the `fd` parameter for future reference. + let idx = insert_or_append(our_sockets, handle) as u16; + + let body = msg.body.memory_message_mut().unwrap(); + let bfr = body.buf.as_slice_mut::(); + log::trace!("successfully connected: {}", idx); + bfr[0] = 0; + bfr[1] = idx; +} + +pub(crate) fn std_udp_rx( + mut msg: xous::MessageEnvelope, + timer: &Ticktimer, + sockets: &mut SocketSet, + udp_rx_waiting: &mut Vec>, + our_sockets: &Vec>, +) { + let connection_handle_index = (msg.body.id() >> 16) & 0xffff; + let body = match msg.body.memory_message_mut() { + Some(body) => body, + None => { + udp_rx_failure(msg, NetError::LibraryError); + return; + } + }; + + // this is not used + body.valid = None; + + let handle = match our_sockets.get(connection_handle_index) { + Some(Some(val)) => val, + _ => { + udp_rx_failure(msg, NetError::Invalid); + return; + } + }; + + let args = body.buf.as_slice::(); + let nonblocking = args[0] == 0; + let expiry = if !nonblocking { + let to = u64::from_le_bytes(args[1..9].try_into().unwrap()); + if to == 0 { + None + } else { + Some(to + timer.elapsed_ms()) + } + } else { + None + }; + + let mut socket = sockets.get::(*handle); + if socket.can_recv() { + log::trace!("receiving data right away"); + match socket.recv() { + Ok((data, endpoint)) => { + udp_rx_success(body.buf.as_slice_mut(), data, endpoint); + } + Err(e) => { + log::error!("unable to receive: {:?}", e); + udp_rx_failure(msg, NetError::LibraryError); + } + } + return; + } + if nonblocking { + udp_rx_failure(msg, NetError::WouldBlock); + return; + } + log::trace!("UDP socket was not ready to receive, adding it to list of waiting messages"); + + // Adding the message to the udp_rx_waiting list prevents it from going out of scope and + // thus prevents the .drop() method from being called. Since messages are returned to the sender + // in the .drop() method, this keeps the caller blocked for the lifetime of the message. + insert_or_append( + udp_rx_waiting, + UdpStdState { + msg, // <-- msg is inserted into the udp_rx_waiting vector, thus preventing the lend_mut from returning. + handle: *handle, + expiry, + }, + ); +} + +pub(crate) fn udp_rx_success(buf: &mut [u8], rx: &[u8], ep: IpEndpoint) { + buf[0] = 0; + let rx_len = (rx.len() as u16).to_le_bytes(); + buf[1] = rx_len[0]; + buf[2] = rx_len[1]; + match ep.addr { + IpAddress::Ipv4(a) => { + buf[3] = 4; + for (&s, d) in a.0.iter().zip(buf[4..8].iter_mut()) { + *d = s; + } + } + IpAddress::Ipv6(a) => { + buf[3] = 6; + for (&s, d) in a.0.iter().zip(buf[4..20].iter_mut()) { + *d = s; + } + } + _ => { + buf[3] = 0; // this is the invalid/error type + } + } + let port = ep.port.to_le_bytes(); + buf[20] = port[0]; + buf[21] = port[1]; + for (&s, d) in rx.iter().zip(buf[22..].iter_mut()) { + *d = s; + } +} + +pub(crate) fn udp_rx_failure(mut env: xous::MessageEnvelope, code: NetError) -> Option<()> { + // If it's not a memory message, don't fill in the return information. + let body = match env.body.memory_message_mut() { + None => { + // But do respond to the scalar message, if it's a BlockingScalar + if env.body.scalar_message().is_some() && env.body.is_blocking() { + xous::return_scalar(env.sender, code as usize).ok(); + } + return None; + } + Some(b) => b, + }; + + body.valid = None; + let s: &mut [u8] = body.buf.as_slice_mut(); + let mut i = s.iter_mut(); + + *i.next()? = 1; + *i.next()? = code as u8; + None +} \ No newline at end of file diff --git a/services/shellchat/src/cmds/net_cmd.rs b/services/shellchat/src/cmds/net_cmd.rs index 7846e98ed..d8ebdc2f5 100644 --- a/services/shellchat/src/cmds/net_cmd.rs +++ b/services/shellchat/src/cmds/net_cmd.rs @@ -1,4 +1,5 @@ use crate::{ShellCmdApi, CommonEnv}; +use com::api::NET_MTU; use xous_ipc::String; #[cfg(any(target_os = "none", target_os = "xous"))] use net::XousServerId; @@ -197,6 +198,39 @@ impl<'a> ShellCmdApi<'a> for NetCmd { }); write!(ret, "TCP listener started on port 80").unwrap(); } + "udpstd" => { + let socket = if let Some(tok_str) = tokens.next() { + tok_str + } else { + "127.0.0.1:6502" + }.to_string(); + let _ = std::thread::spawn({ + let self_cid = self.callback_conn; + move || { + use std::net::UdpSocket; + let udp = match UdpSocket::bind(socket) { + Ok(udp) => udp, + Err(e) => { + log::error!("Couldn't bind UDP socket: {:?}", e); + return; + } + }; + loop { + let mut buf = [0u8; NET_MTU]; + match udp.recv_from(&mut buf) { + Ok((bytes, addr)) => { + let mut s = xous_ipc::String::<512>::new(); + write!(s, "UDP rx {} bytes: {:?}: {}", bytes, addr, std::str::from_utf8(&buf[..bytes]).unwrap()).unwrap(); + s.send(self_cid).unwrap(); + } + Err(e) => { + log::error!("UDP rx failed with {:?}", e); + } + } + } + } + }); + } // Testing of udp is done with netcat: // to send packets run `netcat -u 6502` on a remote host, and then type some data // to receive packets, use `netcat -u -l 6502`, on the same remote host, and it should show a packet of counts received