From 57339dfe2f15f18ae3875e1c601fbf75593ac930 Mon Sep 17 00:00:00 2001 From: bunnie Date: Mon, 4 Apr 2022 22:37:01 +0800 Subject: [PATCH] udp libstd passing inital tests --- services/net/src/api.rs | 60 ++++++++--- services/net/src/main.rs | 132 ++++++++++++++--------- services/net/src/protocols/udp.rs | 5 +- services/net/src/std_udp.rs | 90 +++++++++++++--- services/shellchat/src/cmds/net_cmd.rs | 139 +++++++++++-------------- 5 files changed, 270 insertions(+), 156 deletions(-) diff --git a/services/net/src/api.rs b/services/net/src/api.rs index 2f5904644..2d71492b0 100644 --- a/services/net/src/api.rs +++ b/services/net/src/api.rs @@ -35,12 +35,12 @@ pub const MIN_EC_REV: u32 = 0x00_09_06_00; #[derive(num_derive::FromPrimitive, num_derive::ToPrimitive, Debug)] #[repr(C)] pub(crate) enum Opcode { - /// Calls for UDP implementation + /// Calls for UDP implementation (now deprecated to libstd) UdpBind = 0, UdpClose = 1, UdpTx = 2, - UdpSetTtl = 3, - UdpGetTtl = 4, + //UdpSetTtl = 3, + //UdpGetTtl = 4, /// Calls for TCP implementation TcpConnect = 5, @@ -172,6 +172,7 @@ pub(crate) enum Opcode { StdGetTtl = 36, /// BlockingScalar call to set the maximum hop count of this connection + /// arg4: 0 => TCP, 1 => UDP StdSetTtl = 37, /// BlockingScalar call to get the NODELAY / "Nagle" value of this connection @@ -209,18 +210,19 @@ pub(crate) enum Opcode { /// offset | Contents /// =======|========= /// 0 | 1 indicating error - /// 1 | 1 indicating error (duplicated) - /// 2 | 1 indicating error (duplicated) - /// 3 | 1 indicating error (duplicated) - /// 4 | Error code + /// 1 | Error code StdUdpBind = 40, - //StdUdpClose = 41, + /// Close the UDP connection. The connection ID is specified in the upper 16 bits + /// of the opcode. This may be any kind of message (scalar, blockingscalar, memory, + /// etc.) + StdUdpClose = 41, - /// Receives data from the specified TCP Connection. + /// Receives data from the specified UDP Connection. /// - /// - The UDP connection number is passed in the upper 16 bits of the opcode. - /// - The arguments are passed as follows: + /// - The UDP connection number (fd) is passed in the upper 16 bits of the opcode. + /// # Argumesnts + /// The arguments are passed as follows: /// u8 array /// -------|--------- /// offset | Contents @@ -229,7 +231,8 @@ pub(crate) enum Opcode { /// 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: + /// # Returns + /// The return buffer is the `buf` parameter, which is split as follows: /// u8 array /// -------|--------- /// offset | Contents @@ -251,9 +254,36 @@ pub(crate) enum Opcode { /// - The `valid` and `offset` parameters are not used. StdUdpRx = 42, - //StdUdpTx = 43, - //StdUdpSetTtl = 44, - //StdUdpGetTtl = 45, + /// Send UDP data to a remote host + /// + /// - The UDP connection number (fd) is passed in the upper 16 bits of the opcode. + /// - The `offset` parameter is not used. + /// - The `valid` parameter is not used. + /// # Arguments + /// u8 array + /// -------|--------- + /// offset | Contents + /// =======|========= + /// 0 | port:u16, low byte + /// 1 | port:u16, high byte + /// 2 | address type -- 4 = ipv4, 6 = ipv6, 0 = invalid/err + /// 3 | LSB ip address + /// ... + /// 6 | MSB ipv4 address + /// ... + /// 18| MSW ipv6 address + /// 19| len:u16, low byte + /// 20| len:u16, high byte + /// 21| beginning of &data[..len] to transmit + /// # Returns + /// + /// u8 array + /// -------|--------- + /// offset | Contents + /// =======|========= + /// 0 | 1 indicating error, 0 indicating success + /// 1 | Error code (only valid on error) + StdUdpTx = 43, } #[derive(Debug, Archive, Serialize, Deserialize, Copy, Clone, Default)] diff --git a/services/net/src/main.rs b/services/net/src/main.rs index 90204c442..daff6daff 100644 --- a/services/net/src/main.rs +++ b/services/net/src/main.rs @@ -593,10 +593,17 @@ fn xmain() -> ! { .or_default() .get_mut(connection_idx) { - let socket = sockets.get::(*connection); + let args = msg.body.scalar_message().unwrap(); + let limit = if args.arg4 == 1 { + let socket = sockets.get::(*connection); + socket.hop_limit().unwrap_or(64) as usize + } else { + let socket = sockets.get::(*connection); + socket.hop_limit().unwrap_or(64) as usize + }; xous::return_scalar( msg.sender, - socket.hop_limit().unwrap_or_default() as usize, + limit, ) .ok(); } else { @@ -619,14 +626,19 @@ fn xmain() -> ! { .or_default() .get_mut(connection_idx) { - let mut socket = sockets.get::(*connection); let args = msg.body.scalar_message().unwrap(); - let hop_limit = if args.arg1 == 0 { + let hop_limit = if (args.arg1 == 0) || (args.arg1 > 255) { None } else { Some(args.arg1 as u8) }; - socket.set_hop_limit(hop_limit); + if args.arg4 == 1 { + let mut socket = sockets.get::(*connection); + socket.set_hop_limit(hop_limit); + } else { + let mut socket = sockets.get::(*connection); + socket.set_hop_limit(hop_limit); + } xous::return_scalar(msg.sender, 0).ok(); } else { respond_with_error(msg, NetError::Invalid); @@ -1021,6 +1033,42 @@ fn xmain() -> ! { ); } + Some(Opcode::StdUdpTx) => { + let pid = msg.sender.pid(); + std_udp_tx( + msg, + &mut sockets, + process_sockets.entry(pid).or_default(), + ); + } + + Some(Opcode::StdUdpClose) => { + let pid = msg.sender.pid(); + let connection_idx = msg.body.id() >> 16; + let handle = if let Some(connection) = process_sockets + .entry(pid) + .or_default() + .get_mut(connection_idx) + { + if let Some(connection) = connection.take() { + connection + } else { + udp_failure(msg, NetError::Invalid); + continue; + } + } else { + udp_failure(msg, NetError::Invalid); + continue; + }; + sockets.get::(handle).close(); + sockets.remove(handle); + if let Some(response) = msg.body.memory_message_mut() { + response.buf.as_slice_mut::()[0] = 0; + } else if !msg.body.is_blocking() && msg.body.is_blocking() { + xous::return_scalar(msg.sender, 0).ok(); + } + } + Some(Opcode::UdpBind) => { let mut buf = unsafe { Buffer::from_memory_message_mut(msg.body.memory_message_mut().unwrap()) @@ -1173,38 +1221,6 @@ fn xmain() -> ! { _ => buf.replace(NetMemResponse::Invalid).unwrap(), } } - Some(Opcode::UdpSetTtl) => msg_scalar_unpack!(msg, ttl, port, _, _, { - match udp_handles.get_mut(&(port as u16)) { - Some(udpstate) => { - let mut socket = sockets.get::(udpstate.handle); - let checked_ttl = if ttl > 255 || ttl == 0 { 64 } else { ttl as u8 }; - socket.set_hop_limit(Some(checked_ttl)); - } - None => { - log::error!( - "Set TTL message received, but no port was bound! port {} ttl {}", - port, - ttl - ); - } - } - }), - Some(Opcode::UdpGetTtl) => msg_blocking_scalar_unpack!(msg, port, _, _, _, { - match udp_handles.get_mut(&(port as u16)) { - Some(udpstate) => { - let socket = sockets.get::(udpstate.handle); - let ttl = socket.hop_limit().unwrap_or(64); // 64 is the value used by smoltcp if hop limit isn't set - xous::return_scalar(msg.sender, ttl as usize).expect("couldn't return TTL"); - } - None => { - log::error!( - "Set TTL message received, but no port was bound! port {}", - port - ); - xous::return_scalar(msg.sender, usize::MAX).expect("couldn't return TTL"); - } - } - }), Some(Opcode::ComInterrupt) => { com_int_list.clear(); @@ -1712,23 +1728,41 @@ fn xmain() -> ! { // If it can't receive, then the only explanation was that it timed out if !socket.can_recv() { - udp_rx_failure(msg, NetError::TimedOut); + udp_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 - ); + let do_peek = msg.body.memory_message().unwrap().offset.is_some(); + if do_peek { + match socket.peek() { + 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 // have to duplicate the code between peek and recv because of this type difference + ); + } + Err(e) => { + log::error!("unable to receive: {:?}", e); + udp_failure(msg, NetError::LibraryError); + } } - Err(e) => { - log::error!("unable to receive: {:?}", e); - udp_rx_failure(msg, NetError::LibraryError); + } else { + 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_failure(msg, NetError::LibraryError); + } } } } diff --git a/services/net/src/protocols/udp.rs b/services/net/src/protocols/udp.rs index c7d7a5f57..ee6669402 100644 --- a/services/net/src/protocols/udp.rs +++ b/services/net/src/protocols/udp.rs @@ -13,7 +13,7 @@ use smoltcp::{ time::Duration, }; -use xous::{CID, Message, SID, msg_blocking_scalar_unpack, send_message}; +use xous::{CID, Message, SID, msg_blocking_scalar_unpack}; use xous_ipc::Buffer; use crate::NetConn; use crate::api::*; @@ -362,6 +362,7 @@ impl UdpSocket { Ok(cloned_socket) } + /* pub fn set_ttl(&self, ttl: u32) -> io::Result<()> { if ttl > 255 { return Err(Error::new(ErrorKind::InvalidInput, "TTL must be less than 256")) @@ -383,7 +384,7 @@ impl UdpSocket { Err(Error::new(ErrorKind::ConnectionRefused, "can't get TTL value from Net server")) } } - + */ pub fn take_error(&self) -> io::Result> { // we don't have stateful errors, yet?... Ok(None) diff --git a/services/net/src/std_udp.rs b/services/net/src/std_udp.rs index a8f925a09..a3499766f 100644 --- a/services/net/src/std_udp.rs +++ b/services/net/src/std_udp.rs @@ -19,7 +19,7 @@ pub(crate) fn std_udp_bind( Some(b) => b, None => { log::trace!("invalid message type"); - respond_with_error(msg, NetError::LibraryError); + udp_failure(msg, NetError::LibraryError); return; } }; @@ -30,7 +30,7 @@ pub(crate) fn std_udp_bind( Some(addr) => addr, None => { log::trace!("couldn't parse address"); - respond_with_error(msg, NetError::LibraryError); + udp_failure(msg, NetError::LibraryError); return; } }; @@ -51,7 +51,7 @@ pub(crate) fn std_udp_bind( }) { log::trace!("couldn't connect: {:?}", e); - respond_with_error(msg, e); + udp_failure(msg, e); return; } @@ -78,7 +78,7 @@ pub(crate) fn std_udp_rx( let body = match msg.body.memory_message_mut() { Some(body) => body, None => { - udp_rx_failure(msg, NetError::LibraryError); + udp_failure(msg, NetError::LibraryError); return; } }; @@ -89,7 +89,7 @@ pub(crate) fn std_udp_rx( let handle = match our_sockets.get(connection_handle_index) { Some(Some(val)) => val, _ => { - udp_rx_failure(msg, NetError::Invalid); + udp_failure(msg, NetError::Invalid); return; } }; @@ -106,23 +106,38 @@ pub(crate) fn std_udp_rx( } else { None }; + let do_peek = body.offset.is_some(); 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); + if do_peek { + // have to duplicate the code because Endpoint on peek is &, but on recv is not. This + // difference in types means you can't do a pattern match assign to a common variable. + match socket.peek() { + Ok((data, endpoint)) => { + udp_rx_success(body.buf.as_slice_mut(), data, *endpoint); + } + Err(e) => { + log::error!("unable to receive: {:?}", e); + udp_failure(msg, NetError::LibraryError); + } } - Err(e) => { - log::error!("unable to receive: {:?}", e); - udp_rx_failure(msg, NetError::LibraryError); + } else { + 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_failure(msg, NetError::LibraryError); + } } - } + }; return; } if nonblocking { - udp_rx_failure(msg, NetError::WouldBlock); + udp_failure(msg, NetError::WouldBlock); return; } log::trace!("UDP socket was not ready to receive, adding it to list of waiting messages"); @@ -140,6 +155,53 @@ pub(crate) fn std_udp_rx( ); } +pub(crate) fn std_udp_tx( + mut msg: xous::MessageEnvelope, + sockets: &mut SocketSet, + our_sockets: &Vec>, +) { + // unpack meta + let connection_handle_index = (msg.body.id() >> 16) & 0xffff; + let body = match msg.body.memory_message_mut() { + Some(body) => body, + None => { + udp_failure(msg, NetError::LibraryError); + return; + } + }; + let handle = match our_sockets.get(connection_handle_index) { + Some(Some(val)) => val, + _ => { + udp_failure(msg, NetError::Invalid); + return; + } + }; + + // unpack arguments + let bytes = body.buf.as_slice::(); + let remote_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"); + udp_failure(msg, NetError::LibraryError); + return; + } + }; + let len = u16::from_le_bytes([bytes[19], bytes[20]]); + // attempt the tx + let mut socket = sockets.get::(*handle); + match socket.send_slice(&bytes[21..21 + len as usize], IpEndpoint::new(address, remote_port)) { + Ok(_) => { + body.buf.as_slice_mut()[0] = 0; + } + Err(_e) => { + // the only type of error returned from smoltcp in this case is if the destination is not addressible. + udp_failure(msg, NetError::Unaddressable); + } + } +} + 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(); @@ -170,7 +232,7 @@ pub(crate) fn udp_rx_success(buf: &mut [u8], rx: &[u8], ep: IpEndpoint) { } } -pub(crate) fn udp_rx_failure(mut env: xous::MessageEnvelope, code: NetError) -> Option<()> { +pub(crate) fn udp_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 => { diff --git a/services/shellchat/src/cmds/net_cmd.rs b/services/shellchat/src/cmds/net_cmd.rs index d8ebdc2f5..01f1bddb1 100644 --- a/services/shellchat/src/cmds/net_cmd.rs +++ b/services/shellchat/src/cmds/net_cmd.rs @@ -56,10 +56,10 @@ impl<'a> ShellCmdApi<'a> for NetCmd { use core::fmt::Write; let mut ret = String::<1024>::new(); #[cfg(any(target_os = "none", target_os = "xous"))] - let helpstring = "net [udp [port]] [udpclose] [udpclone] [udpcloneclose] [ping [host] [count]] [tcpget host/path]"; + let helpstring = "net [udp [rx socket] [tx dest socket]] [ping [host] [count]] [tcpget host/path]"; // no ping in hosted mode -- why would you need it? we're using the host's network connection. #[cfg(not(any(target_os = "none", target_os = "xous")))] - let helpstring = "net [udp [port]] [udpclose] [udpclone] [udpcloneclose] [count]] [tcpget host/path]"; + let helpstring = "net [udp [port]] [count]] [tcpget host/path]"; let mut tokens = args.as_str().unwrap().split(' '); @@ -198,87 +198,74 @@ 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 "udp" => { - if let Some(udp_socket) = &self.udp { - write!(ret, "Socket listener already installed on {:?}.", udp_socket.socket_addr().unwrap()).unwrap(); + let socket = if let Some(tok_str) = tokens.next() { + tok_str } else { - let socket = if let Some(tok_str) = tokens.next() { - tok_str - } else { - "127.0.0.1:6502" - }; - let mut udp = net::UdpSocket::bind_xous( - socket, - Some(UDP_TEST_SIZE as u16) - ).unwrap(); - udp.set_read_timeout(Some(Duration::from_millis(1000))).unwrap(); - udp.set_scalar_notification( - self.callback_conn, - self.callback_id.unwrap() as usize, // this is guaranteed in the prelude - [Some(NetCmdDispatch::UdpTest1.to_usize().unwrap()), None, None, None] - ); - self.udp = Some(udp); - write!(ret, "Created UDP socket listener on socket {}", socket).unwrap(); - } - } - "udpclose" => { - self.udp = None; - write!(ret, "Closed primary UDP socket").unwrap(); - } - "udpclone" => { - if let Some(udp_socket) = &self.udp { - let mut udp_clone = udp_socket.duplicate().unwrap(); - udp_clone.set_scalar_notification( - self.callback_conn, - self.callback_id.unwrap() as usize, // this is guaranteed in the prelude - [Some(NetCmdDispatch::UdpTest2.to_usize().unwrap()), None, None, None] - ); - let sa = udp_clone.socket_addr().unwrap(); - self.udp_clone = Some(udp_clone); - write!(ret, "Cloned UDP socket on {:?}", sa).unwrap(); + write!(ret, "Usage: net udp precursor_ip:6502 [sender_ip]:6502, where precursor_ip is the IP address of the device itself").unwrap(); + return Ok(Some(ret)); + }.to_string(); + let (response_addr, do_response) = if let Some(r) = tokens.next() { + (r.to_string(), true) } else { - write!(ret, "Run `net udp` before cloning.").unwrap(); + (std::string::String::new(), false) + }; + use std::net::UdpSocket; + let udp = match UdpSocket::bind(socket.clone()) { + Ok(udp) => udp, + Err(e) => { + write!(ret, "Couldn't bind UDP socket: {:?}\n", e).unwrap(); + return Ok(Some(ret)); + } + }; + udp.set_write_timeout(Some(std::time::Duration::from_millis(500))).expect("couldn't set write timeout"); + for index in 0..2 { + let _ = std::thread::spawn({ + let self_cid = self.callback_conn; + let udp = udp.try_clone().expect("couldn't clone socket"); + let response = response_addr.clone(); + move || { + const ITERS: usize = 4; + let mut iters = 0; + let mut s = xous_ipc::String::<512>::new(); + write!(s, "UDP server {} started", index).unwrap(); + s.send(self_cid).unwrap(); + loop { + s.clear(); + let mut buf = [0u8; NET_MTU]; + match udp.recv_from(&mut buf) { + Ok((bytes, addr)) => { + write!(s, "UDP server {} rx {} bytes: {:?}: {}\r\n", index, bytes, addr, std::str::from_utf8(&buf[..bytes]).unwrap()).unwrap(); + s.send(self_cid).unwrap(); + if do_response { + match udp.send_to( + format!("Server {} received {} bytes", index, bytes).as_bytes(), + &response, + ) { + Ok(len) => log::info!("server {} sent response of {} bytes", index, len), + Err(e) => log::info!("server {} UDP tx err: {:?}", index, e), + } + } + } + Err(e) => { + log::error!("UDP rx failed with {:?}", e); + } + } + iters += 1; + if iters >= ITERS { + break; + } + } + s.clear(); + write!(s, "UDP server {} rx closed after {} iters", index, iters).unwrap(); + s.send(self_cid).unwrap(); + } + }); } - } - "udpcloneclose" => { - self.udp_clone = None; - write!(ret, "Closed cloned UDP socket").unwrap(); + write!(ret, "Started multi-threaded UDP responder").unwrap(); } "dns" => { if let Some(name) = tokens.next() {