saymrwulf/xous-core
1
0
Fork 0
mirror of https://github.com/saymrwulf/xous-core.git synced 2026-11-08 04:56:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

first draft at some basic libstd UDP mechanics

Browse source
This commit is contained in:
bunnie 2022-04-04 04:49:17 +08:00
parent 9dc75071ad
commit b5a6c436c2
5 changed files with 400 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
services/net/src/api.rs
View file

@ -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

89
services/net/src/main.rs
View file

@ -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<NonZeroU64>,
}
pub struct UdpStdState {
pub msg: xous::MessageEnvelope,
pub handle: SocketHandle,
pub expiry: Option<u64>,
}
fn set_com_ints(com_int_list: &mut Vec<ComIntSources>) {
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<Option<xous::PID>, Vec<Option<SocketHandle>>> = 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<Option<WaitingSocket>> = 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<Option<UdpStdState>> = 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::<UdpSocket>(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::<IcmpSocket>(icmp_handle);

4
services/net/src/std_glue.rs
View file

@ -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;

193
services/net/src/std_udp.rs Normal file
View file

@ -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<Option<SocketHandle>>,
) {
// 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::<u8>();
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::<u16>();
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<Option<UdpStdState>>,
our_sockets: &Vec<Option<SocketHandle>>,
) {
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::<u8>();
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::<UdpSocket>(*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
}

34
services/shellchat/src/cmds/net_cmd.rs
View file

@ -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 <precursor ip address> 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