Merge #1226
1226: embassy-net: Implement flush for TcpSocket r=Dirbaio a=kbleeke Implements flush for TcpSocket by checking the send queue. Flushing is implemented by checking if smoltcp's send_queue/tx_buffer is empty. The flush is completed when all outstanding octets are acknowledged. Smoltcp wakes the send waker [here](https://docs.rs/smoltcp/latest/src/smoltcp/socket/tcp.rs.html#1712) when ACKs are processed and data is removed from the send buffer. So we can re-check in our flush implementation, if the buffer is now empty. fixes #1223 Co-authored-by: kbleeke <pluth@0t.re>
This commit is contained in:
commit
464faa2a04
1 changed files with 20 additions and 3 deletions
|
@ -63,6 +63,10 @@ impl<'a> TcpWriter<'a> {
|
||||||
pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
|
pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
|
||||||
self.io.write(buf).await
|
self.io.write(buf).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub async fn flush(&mut self) -> Result<(), Error> {
|
||||||
|
self.io.flush().await
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<'a> TcpSocket<'a> {
|
impl<'a> TcpSocket<'a> {
|
||||||
|
@ -146,6 +150,10 @@ impl<'a> TcpSocket<'a> {
|
||||||
self.io.write(buf).await
|
self.io.write(buf).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub async fn flush(&mut self) -> Result<(), Error> {
|
||||||
|
self.io.flush().await
|
||||||
|
}
|
||||||
|
|
||||||
pub fn set_timeout(&mut self, duration: Option<Duration>) {
|
pub fn set_timeout(&mut self, duration: Option<Duration>) {
|
||||||
self.io.with_mut(|s, _| s.set_timeout(duration))
|
self.io.with_mut(|s, _| s.set_timeout(duration))
|
||||||
}
|
}
|
||||||
|
@ -254,10 +262,19 @@ impl<'d> TcpIo<'d> {
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
#[allow(unused)]
|
|
||||||
async fn flush(&mut self) -> Result<(), Error> {
|
async fn flush(&mut self) -> Result<(), Error> {
|
||||||
poll_fn(move |_| {
|
poll_fn(move |cx| {
|
||||||
Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
|
self.with_mut(|s, _| {
|
||||||
|
// If there are outstanding send operations, register for wake up and wait
|
||||||
|
// smoltcp issues wake-ups when octets are dequeued from the send buffer
|
||||||
|
if s.send_queue() > 0 {
|
||||||
|
s.register_send_waker(cx.waker());
|
||||||
|
Poll::Pending
|
||||||
|
// No outstanding sends, socket is flushed
|
||||||
|
} else {
|
||||||
|
Poll::Ready(Ok(()))
|
||||||
|
}
|
||||||
|
})
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue