max/embassy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP embassy-net v2

Browse Source
This commit is contained in:
Dario Nieuwenhuis
2022-05-23 03:50:43 +02:00
parent 36a1f20364
commit a5aea995a8
17 changed files with 660 additions and 734 deletions
Download Patch File Download Diff File Expand all files Collapse all files

305
embassy-net/src/tcp.rs
View File

@ -1,13 +1,16 @@
use core::cell::UnsafeCell;
use core::future::Future;
use core::marker::PhantomData;
use core::mem;
use core::task::Poll;
use futures::future::poll_fn;
use smoltcp::iface::{Context as SmolContext, SocketHandle};
use smoltcp::socket::TcpSocket as SyncTcpSocket;
use smoltcp::socket::{TcpSocketBuffer, TcpState};
use smoltcp::iface::{Interface, SocketHandle};
use smoltcp::socket::tcp;
use smoltcp::time::Duration;
use smoltcp::wire::IpEndpoint;
use smoltcp::wire::IpListenEndpoint;
use crate::stack::SocketStack;
use crate::Device;
use super::stack::Stack;
@ -42,78 +45,68 @@ pub enum AcceptError {
}
pub struct TcpSocket<'a> {
handle: SocketHandle,
ghost: PhantomData<&'a mut [u8]>,
io: TcpIo<'a>,
}
impl<'a> Unpin for TcpSocket<'a> {}
pub struct TcpReader<'a> {
handle: SocketHandle,
ghost: PhantomData<&'a mut [u8]>,
io: TcpIo<'a>,
}
impl<'a> Unpin for TcpReader<'a> {}
pub struct TcpWriter<'a> {
handle: SocketHandle,
ghost: PhantomData<&'a mut [u8]>,
io: TcpIo<'a>,
}
impl<'a> Unpin for TcpWriter<'a> {}
impl<'a> TcpSocket<'a> {
pub fn new(rx_buffer: &'a mut [u8], tx_buffer: &'a mut [u8]) -> Self {
let handle = Stack::with(|stack| {
let rx_buffer: &'static mut [u8] = unsafe { mem::transmute(rx_buffer) };
let tx_buffer: &'static mut [u8] = unsafe { mem::transmute(tx_buffer) };
stack.iface.add_socket(SyncTcpSocket::new(
TcpSocketBuffer::new(rx_buffer),
TcpSocketBuffer::new(tx_buffer),
))
});
pub fn new<D: Device>(
stack: &'a Stack<D>,
rx_buffer: &'a mut [u8],
tx_buffer: &'a mut [u8],
) -> Self {
// safety: not accessed reentrantly.
let s = unsafe { &mut *stack.socket.get() };
let rx_buffer: &'static mut [u8] = unsafe { mem::transmute(rx_buffer) };
let tx_buffer: &'static mut [u8] = unsafe { mem::transmute(tx_buffer) };
let handle = s.sockets.add(tcp::Socket::new(
tcp::SocketBuffer::new(rx_buffer),
tcp::SocketBuffer::new(tx_buffer),
));
Self {
handle,
ghost: PhantomData,
io: TcpIo {
stack: &stack.socket,
handle,
},
}
}
pub fn split(&mut self) -> (TcpReader<'_>, TcpWriter<'_>) {
(
TcpReader {
handle: self.handle,
ghost: PhantomData,
},
TcpWriter {
handle: self.handle,
ghost: PhantomData,
},
)
(TcpReader { io: self.io }, TcpWriter { io: self.io })
}
pub async fn connect<T>(&mut self, remote_endpoint: T) -> Result<(), ConnectError>
where
T: Into<IpEndpoint>,
{
let local_port = Stack::with(|stack| stack.get_local_port());
match with_socket(self.handle, |s, cx| {
s.connect(cx, remote_endpoint, local_port)
}) {
// safety: not accessed reentrantly.
let local_port = unsafe { &mut *self.io.stack.get() }.get_local_port();
// safety: not accessed reentrantly.
match unsafe {
self.io
.with_mut(|s, i| s.connect(i, remote_endpoint, local_port))
} {
Ok(()) => {}
Err(smoltcp::Error::Illegal) => return Err(ConnectError::InvalidState),
Err(smoltcp::Error::Unaddressable) => return Err(ConnectError::NoRoute),
// smoltcp returns no errors other than the above.
Err(_) => unreachable!(),
Err(tcp::ConnectError::InvalidState) => return Err(ConnectError::InvalidState),
Err(tcp::ConnectError::Unaddressable) => return Err(ConnectError::NoRoute),
}
futures::future::poll_fn(|cx| {
with_socket(self.handle, |s, _| match s.state() {
TcpState::Closed | TcpState::TimeWait => {
futures::future::poll_fn(|cx| unsafe {
self.io.with_mut(|s, _| match s.state() {
tcp::State::Closed | tcp::State::TimeWait => {
Poll::Ready(Err(ConnectError::ConnectionReset))
}
TcpState::Listen => unreachable!(),
TcpState::SynSent | TcpState::SynReceived => {
tcp::State::Listen => unreachable!(),
tcp::State::SynSent | tcp::State::SynReceived => {
s.register_send_waker(cx.waker());
Poll::Pending
}
@ -125,19 +118,18 @@ impl<'a> TcpSocket<'a> {
pub async fn accept<T>(&mut self, local_endpoint: T) -> Result<(), AcceptError>
where
T: Into<IpEndpoint>,
T: Into<IpListenEndpoint>,
{
match with_socket(self.handle, |s, _| s.listen(local_endpoint)) {
// safety: not accessed reentrantly.
match unsafe { self.io.with_mut(|s, _| s.listen(local_endpoint)) } {
Ok(()) => {}
Err(smoltcp::Error::Illegal) => return Err(AcceptError::InvalidState),
Err(smoltcp::Error::Unaddressable) => return Err(AcceptError::InvalidPort),
// smoltcp returns no errors other than the above.
Err(_) => unreachable!(),
Err(tcp::ListenError::InvalidState) => return Err(AcceptError::InvalidState),
Err(tcp::ListenError::Unaddressable) => return Err(AcceptError::InvalidPort),
}
futures::future::poll_fn(|cx| {
with_socket(self.handle, |s, _| match s.state() {
TcpState::Listen | TcpState::SynSent | TcpState::SynReceived => {
futures::future::poll_fn(|cx| unsafe {
self.io.with_mut(|s, _| match s.state() {
tcp::State::Listen | tcp::State::SynSent | tcp::State::SynReceived => {
s.register_send_waker(cx.waker());
Poll::Pending
}
@ -148,65 +140,122 @@ impl<'a> TcpSocket<'a> {
}
pub fn set_timeout(&mut self, duration: Option<Duration>) {
with_socket(self.handle, |s, _| s.set_timeout(duration))
unsafe { self.io.with_mut(|s, _| s.set_timeout(duration)) }
}
pub fn set_keep_alive(&mut self, interval: Option<Duration>) {
with_socket(self.handle, |s, _| s.set_keep_alive(interval))
unsafe { self.io.with_mut(|s, _| s.set_keep_alive(interval)) }
}
pub fn set_hop_limit(&mut self, hop_limit: Option<u8>) {
with_socket(self.handle, |s, _| s.set_hop_limit(hop_limit))
unsafe { self.io.with_mut(|s, _| s.set_hop_limit(hop_limit)) }
}
pub fn local_endpoint(&self) -> IpEndpoint {
with_socket(self.handle, |s, _| s.local_endpoint())
pub fn local_endpoint(&self) -> Option<IpEndpoint> {
unsafe { self.io.with(|s, _| s.local_endpoint()) }
}
pub fn remote_endpoint(&self) -> IpEndpoint {
with_socket(self.handle, |s, _| s.remote_endpoint())
pub fn remote_endpoint(&self) -> Option<IpEndpoint> {
unsafe { self.io.with(|s, _| s.remote_endpoint()) }
}
pub fn state(&self) -> TcpState {
with_socket(self.handle, |s, _| s.state())
pub fn state(&self) -> tcp::State {
unsafe { self.io.with(|s, _| s.state()) }
}
pub fn close(&mut self) {
with_socket(self.handle, |s, _| s.close())
unsafe { self.io.with_mut(|s, _| s.close()) }
}
pub fn abort(&mut self) {
with_socket(self.handle, |s, _| s.abort())
unsafe { self.io.with_mut(|s, _| s.abort()) }
}
pub fn may_send(&self) -> bool {
with_socket(self.handle, |s, _| s.may_send())
unsafe { self.io.with(|s, _| s.may_send()) }
}
pub fn may_recv(&self) -> bool {
with_socket(self.handle, |s, _| s.may_recv())
unsafe { self.io.with(|s, _| s.may_recv()) }
}
}
fn with_socket<R>(
handle: SocketHandle,
f: impl FnOnce(&mut SyncTcpSocket, &mut SmolContext) -> R,
) -> R {
Stack::with(|stack| {
let res = {
let (s, cx) = stack.iface.get_socket_and_context::<SyncTcpSocket>(handle);
f(s, cx)
};
stack.wake();
res
})
}
impl<'a> Drop for TcpSocket<'a> {
fn drop(&mut self) {
Stack::with(|stack| {
stack.iface.remove_socket(self.handle);
// safety: not accessed reentrantly.
let s = unsafe { &mut *self.io.stack.get() };
s.sockets.remove(self.io.handle);
}
}
// =======================
#[derive(Copy, Clone)]
pub struct TcpIo<'a> {
stack: &'a UnsafeCell<SocketStack>,
handle: SocketHandle,
}
impl<'d> TcpIo<'d> {
/// SAFETY: must not call reentrantly.
unsafe fn with<R>(&self, f: impl FnOnce(&tcp::Socket, &Interface) -> R) -> R {
let s = &*self.stack.get();
let socket = s.sockets.get::<tcp::Socket>(self.handle);
f(socket, &s.iface)
}
/// SAFETY: must not call reentrantly.
unsafe fn with_mut<R>(&mut self, f: impl FnOnce(&mut tcp::Socket, &mut Interface) -> R) -> R {
let s = &mut *self.stack.get();
let socket = s.sockets.get_mut::<tcp::Socket>(self.handle);
let res = f(socket, &mut s.iface);
s.waker.wake();
res
}
async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Error> {
poll_fn(move |cx| unsafe {
// CAUTION: smoltcp semantics around EOF are different to what you'd expect
// from posix-like IO, so we have to tweak things here.
self.with_mut(|s, _| match s.recv_slice(buf) {
// No data ready
Ok(0) => {
s.register_recv_waker(cx.waker());
Poll::Pending
}
// Data ready!
Ok(n) => Poll::Ready(Ok(n)),
// EOF
Err(tcp::RecvError::Finished) => Poll::Ready(Ok(0)),
// Connection reset. TODO: this can also be timeouts etc, investigate.
Err(tcp::RecvError::InvalidState) => Poll::Ready(Err(Error::ConnectionReset)),
})
})
.await
}
async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
poll_fn(move |cx| unsafe {
self.with_mut(|s, _| match s.send_slice(buf) {
// Not ready to send (no space in the tx buffer)
Ok(0) => {
s.register_send_waker(cx.waker());
Poll::Pending
}
// Some data sent
Ok(n) => Poll::Ready(Ok(n)),
// Connection reset. TODO: this can also be timeouts etc, investigate.
Err(tcp::SendError::InvalidState) => Poll::Ready(Err(Error::ConnectionReset)),
})
})
.await
}
async fn flush(&mut self) -> Result<(), Error> {
poll_fn(move |_| {
Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
})
.await
}
}
@ -226,25 +275,7 @@ impl<'d> embedded_io::asynch::Read for TcpSocket<'d> {
Self: 'a;
fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
poll_fn(move |cx| {
// CAUTION: smoltcp semantics around EOF are different to what you'd expect
// from posix-like IO, so we have to tweak things here.
with_socket(self.handle, |s, _| match s.recv_slice(buf) {
// No data ready
Ok(0) => {
s.register_recv_waker(cx.waker());
Poll::Pending
}
// Data ready!
Ok(n) => Poll::Ready(Ok(n)),
// EOF
Err(smoltcp::Error::Finished) => Poll::Ready(Ok(0)),
// Connection reset. TODO: this can also be timeouts etc, investigate.
Err(smoltcp::Error::Illegal) => Poll::Ready(Err(Error::ConnectionReset)),
// smoltcp returns no errors other than the above.
Err(_) => unreachable!(),
})
})
self.io.read(buf)
}
}
@ -254,21 +285,7 @@ impl<'d> embedded_io::asynch::Write for TcpSocket<'d> {
Self: 'a;
fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
poll_fn(move |cx| {
with_socket(self.handle, |s, _| match s.send_slice(buf) {
// Not ready to send (no space in the tx buffer)
Ok(0) => {
s.register_send_waker(cx.waker());
Poll::Pending
}
// Some data sent
Ok(n) => Poll::Ready(Ok(n)),
// Connection reset. TODO: this can also be timeouts etc, investigate.
Err(smoltcp::Error::Illegal) => Poll::Ready(Err(Error::ConnectionReset)),
// smoltcp returns no errors other than the above.
Err(_) => unreachable!(),
})
})
self.io.write(buf)
}
type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
@ -276,9 +293,7 @@ impl<'d> embedded_io::asynch::Write for TcpSocket<'d> {
Self: 'a;
fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
poll_fn(move |_| {
Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
})
self.io.flush()
}
}
@ -292,25 +307,7 @@ impl<'d> embedded_io::asynch::Read for TcpReader<'d> {
Self: 'a;
fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
poll_fn(move |cx| {
// CAUTION: smoltcp semantics around EOF are different to what you'd expect
// from posix-like IO, so we have to tweak things here.
with_socket(self.handle, |s, _| match s.recv_slice(buf) {
// No data ready
Ok(0) => {
s.register_recv_waker(cx.waker());
Poll::Pending
}
// Data ready!
Ok(n) => Poll::Ready(Ok(n)),
// EOF
Err(smoltcp::Error::Finished) => Poll::Ready(Ok(0)),
// Connection reset. TODO: this can also be timeouts etc, investigate.
Err(smoltcp::Error::Illegal) => Poll::Ready(Err(Error::ConnectionReset)),
// smoltcp returns no errors other than the above.
Err(_) => unreachable!(),
})
})
self.io.read(buf)
}
}
@ -324,21 +321,7 @@ impl<'d> embedded_io::asynch::Write for TcpWriter<'d> {
Self: 'a;
fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
poll_fn(move |cx| {
with_socket(self.handle, |s, _| match s.send_slice(buf) {
// Not ready to send (no space in the tx buffer)
Ok(0) => {
s.register_send_waker(cx.waker());
Poll::Pending
}
// Some data sent
Ok(n) => Poll::Ready(Ok(n)),
// Connection reset. TODO: this can also be timeouts etc, investigate.
Err(smoltcp::Error::Illegal) => Poll::Ready(Err(Error::ConnectionReset)),
// smoltcp returns no errors other than the above.
Err(_) => unreachable!(),
})
})
self.io.write(buf)
}
type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
@ -346,8 +329,6 @@ impl<'d> embedded_io::asynch::Write for TcpWriter<'d> {
Self: 'a;
fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
poll_fn(move |_| {
Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
})
self.io.flush()
}
}