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

Replace embassy::io with embedded_io.

Browse Source
This commit is contained in:
Dario Nieuwenhuis
2022-05-04 20:48:37 +02:00
parent fc32b3750c
commit 931a137f8c
47 changed files with 697 additions and 1760 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
embassy-net/Cargo.toml
View File

@ -31,12 +31,15 @@ pool-32 = []
pool-64 = []
pool-128 = []
nightly = ["embedded-io/async"]
[dependencies]
defmt = { version = "0.3", optional = true }
log = { version = "0.4.14", optional = true }
embassy = { version = "0.1.0", path = "../embassy" }
embedded-io = "0.2.0"
managed = { version = "0.8.0", default-features = false, features = [ "map" ] }
heapless = { version = "0.7.5", default-features = false }

9
embassy-net/src/device.rs
View File

@ -4,7 +4,6 @@ use smoltcp::phy::DeviceCapabilities;
use smoltcp::time::Instant as SmolInstant;
use crate::packet_pool::PacketBoxExt;
use crate::Result;
use crate::{Packet, PacketBox, PacketBuf};
#[derive(PartialEq, Eq, Clone, Copy)]
@ -78,9 +77,9 @@ pub struct RxToken {
}
impl smoltcp::phy::RxToken for RxToken {
fn consume<R, F>(mut self, _timestamp: SmolInstant, f: F) -> Result<R>
fn consume<R, F>(mut self, _timestamp: SmolInstant, f: F) -> smoltcp::Result<R>
where
F: FnOnce(&mut [u8]) -> Result<R>,
F: FnOnce(&mut [u8]) -> smoltcp::Result<R>,
{
f(&mut self.pkt)
}
@ -92,9 +91,9 @@ pub struct TxToken<'a> {
}
impl<'a> smoltcp::phy::TxToken for TxToken<'a> {
fn consume<R, F>(self, _timestamp: SmolInstant, len: usize, f: F) -> Result<R>
fn consume<R, F>(self, _timestamp: SmolInstant, len: usize, f: F) -> smoltcp::Result<R>
where
F: FnOnce(&mut [u8]) -> Result<R>,
F: FnOnce(&mut [u8]) -> smoltcp::Result<R>,
{
let mut buf = self.pkt.slice(0..len);
let r = f(&mut buf)?;

9
embassy-net/src/lib.rs
View File

@ -1,5 +1,9 @@
#![cfg_attr(not(feature = "std"), no_std)]
#![allow(clippy::new_without_default)]
#![cfg_attr(
feature = "nightly",
feature(generic_associated_types, type_alias_impl_trait)
)]
// This mod MUST go first, so that the others see its macros.
pub(crate) mod fmt;
@ -20,9 +24,7 @@ pub use stack::{
};
#[cfg(feature = "tcp")]
mod tcp_socket;
#[cfg(feature = "tcp")]
pub use tcp_socket::TcpSocket;
pub mod tcp;
// smoltcp reexports
pub use smoltcp::phy::{DeviceCapabilities, Medium};
@ -32,4 +34,3 @@ pub use smoltcp::time::Instant as SmolInstant;
pub use smoltcp::wire::{EthernetAddress, HardwareAddress};
pub use smoltcp::wire::{IpAddress, IpCidr, Ipv4Address, Ipv4Cidr};
pub type Interface = smoltcp::iface::Interface<'static, device::DeviceAdapter>;
pub use smoltcp::{Error, Result};

67
embassy-net/src/tcp/io_impl.rs Normal file
View File

@ -0,0 +1,67 @@
use core::future::Future;
use core::task::Poll;
use futures::future::poll_fn;
use super::{Error, TcpSocket};
impl<'d> embedded_io::asynch::Read for TcpSocket<'d> {
type ReadFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
where
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.
self.with(|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!(),
})
})
}
}
impl<'d> embedded_io::asynch::Write for TcpSocket<'d> {
type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
where
Self: 'a;
fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
poll_fn(move |cx| {
self.with(|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!(),
})
})
}
type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
where
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?
})
}
}

131
embassy-net/src/tcp_socket.rs → embassy-net/src/tcp/mod.rs
View File

@ -1,17 +1,46 @@
use core::marker::PhantomData;
use core::mem;
use core::pin::Pin;
use core::task::{Context, Poll};
use embassy::io;
use embassy::io::{AsyncBufRead, AsyncWrite};
use core::task::Poll;
use smoltcp::iface::{Context as SmolContext, SocketHandle};
use smoltcp::socket::TcpSocket as SyncTcpSocket;
use smoltcp::socket::{TcpSocketBuffer, TcpState};
use smoltcp::time::Duration;
use smoltcp::wire::IpEndpoint;
#[cfg(feature = "nightly")]
mod io_impl;
use super::stack::Stack;
use crate::{Error, Result};
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum Error {
ConnectionReset,
}
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum ConnectError {
/// The socket is already connected or listening.
InvalidState,
/// The remote host rejected the connection with a RST packet.
ConnectionReset,
/// Connect timed out.
TimedOut,
/// No route to host.
NoRoute,
}
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum AcceptError {
/// The socket is already connected or listening.
InvalidState,
/// Invalid listen port
InvalidPort,
/// The remote host rejected the connection with a RST packet.
ConnectionReset,
}
pub struct TcpSocket<'a> {
handle: SocketHandle,
@ -37,17 +66,25 @@ impl<'a> TcpSocket<'a> {
}
}
pub async fn connect<T>(&mut self, remote_endpoint: T) -> Result<()>
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());
self.with(|s, cx| s.connect(cx, remote_endpoint, local_port))?;
match self.with(|s, cx| s.connect(cx, 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!(),
}
futures::future::poll_fn(|cx| {
self.with(|s, _| match s.state() {
TcpState::Closed | TcpState::TimeWait => Poll::Ready(Err(Error::Unaddressable)),
TcpState::Listen => Poll::Ready(Err(Error::Illegal)),
TcpState::Closed | TcpState::TimeWait => {
Poll::Ready(Err(ConnectError::ConnectionReset))
}
TcpState::Listen => unreachable!(),
TcpState::SynSent | TcpState::SynReceived => {
s.register_send_waker(cx.waker());
Poll::Pending
@ -58,11 +95,17 @@ impl<'a> TcpSocket<'a> {
.await
}
pub async fn accept<T>(&mut self, local_endpoint: T) -> Result<()>
pub async fn accept<T>(&mut self, local_endpoint: T) -> Result<(), AcceptError>
where
T: Into<IpEndpoint>,
{
self.with(|s, _| s.listen(local_endpoint))?;
match self.with(|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!(),
}
futures::future::poll_fn(|cx| {
self.with(|s, _| match s.state() {
@ -130,11 +173,6 @@ impl<'a> TcpSocket<'a> {
}
}
fn to_ioerr(_err: Error) -> io::Error {
// todo
io::Error::Other
}
impl<'a> Drop for TcpSocket<'a> {
fn drop(&mut self) {
Stack::with(|stack| {
@ -143,63 +181,12 @@ impl<'a> Drop for TcpSocket<'a> {
}
}
impl<'a> AsyncBufRead for TcpSocket<'a> {
fn poll_fill_buf<'z>(
self: Pin<&'z mut Self>,
cx: &mut Context<'_>,
) -> Poll<io::Result<&'z [u8]>> {
self.with(|s, _| match s.peek(1 << 30) {
// No data ready
Ok(buf) if buf.is_empty() => {
s.register_recv_waker(cx.waker());
Poll::Pending
}
// Data ready!
Ok(buf) => {
// Safety:
// - User can't touch the inner TcpSocket directly at all.
// - The socket itself won't touch these bytes until consume() is called, which
// requires the user to release this borrow.
let buf: &'z [u8] = unsafe { core::mem::transmute(&*buf) };
Poll::Ready(Ok(buf))
}
// EOF
Err(Error::Finished) => Poll::Ready(Ok(&[][..])),
// Error
Err(e) => Poll::Ready(Err(to_ioerr(e))),
})
}
fn consume(self: Pin<&mut Self>, amt: usize) {
if amt == 0 {
// smoltcp's recv returns Finished if we're at EOF,
// even if we're "reading" 0 bytes.
return;
}
self.with(|s, _| s.recv(|_| (amt, ()))).unwrap()
impl embedded_io::Error for Error {
fn kind(&self) -> embedded_io::ErrorKind {
embedded_io::ErrorKind::Other
}
}
impl<'a> AsyncWrite for TcpSocket<'a> {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
self.with(|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)),
// Error
Err(e) => Poll::Ready(Err(to_ioerr(e))),
})
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
}
impl<'d> embedded_io::Io for TcpSocket<'d> {
type Error = Error;
}