From 62733a659424d56296b6983fc02deaa14b374e58 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 5 Aug 2019 14:18:18 -0700 Subject: [PATCH] udp: remove poll_* fns in favor of async fns (#1393) This removes the need for manual futures. --- tokio-udp/Cargo.toml | 3 +- tokio-udp/src/lib.rs | 9 +-- tokio-udp/src/recv.rs | 30 -------- tokio-udp/src/recv_from.rs | 31 -------- tokio-udp/src/send.rs | 30 -------- tokio-udp/src/send_to.rs | 40 ----------- tokio-udp/src/socket.rs | 140 ++++--------------------------------- tokio-udp/src/split.rs | 21 +++--- tokio/src/net.rs | 2 +- 9 files changed, 28 insertions(+), 278 deletions(-) delete mode 100644 tokio-udp/src/recv.rs delete mode 100644 tokio-udp/src/recv_from.rs delete mode 100644 tokio-udp/src/send.rs delete mode 100644 tokio-udp/src/send_to.rs diff --git a/tokio-udp/Cargo.toml b/tokio-udp/Cargo.toml index 2af96b5d2..2c54cf887 100644 --- a/tokio-udp/Cargo.toml +++ b/tokio-udp/Cargo.toml @@ -28,7 +28,8 @@ tokio-reactor = { version = "0.2.0", path = "../tokio-reactor" } # bytes = "0.4" mio = "0.6.14" log = "0.4" -futures-core-preview = "0.3.0-alpha.17" +futures-core-preview = "= 0.3.0-alpha.17" +futures-util-preview = "= 0.3.0-alpha.17" [dev-dependencies] env_logger = { version = "0.5", default-features = false } diff --git a/tokio-udp/src/lib.rs b/tokio-udp/src/lib.rs index 13ffeeec9..a531f2f12 100644 --- a/tokio-udp/src/lib.rs +++ b/tokio-udp/src/lib.rs @@ -2,6 +2,7 @@ #![deny(missing_docs, missing_debug_implementations, rust_2018_idioms)] #![cfg_attr(test, deny(warnings))] #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))] +#![feature(async_await)] //! UDP bindings for `tokio`. //! @@ -13,17 +14,9 @@ //! [`Recv`], [`Send`], [`RecvFrom`] and [`SendTo`] structs respectively. // mod frame; -mod recv; -mod recv_from; -mod send; -mod send_to; mod socket; pub mod split; // pub use self::frame::UdpFramed; -pub use self::recv::Recv; -pub use self::recv_from::RecvFrom; -pub use self::send::Send; -pub use self::send_to::SendTo; pub use self::socket::UdpSocket; diff --git a/tokio-udp/src/recv.rs b/tokio-udp/src/recv.rs deleted file mode 100644 index 47331d3ab..000000000 --- a/tokio-udp/src/recv.rs +++ /dev/null @@ -1,30 +0,0 @@ -use super::UdpSocket; -use std::future::Future; -use std::io; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that receives a datagram from the connected address. -/// -/// This `struct` is created by [`recv`](super::UdpSocket::recv). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct Recv<'a, 'b> { - socket: &'a UdpSocket, - buf: &'b mut [u8], -} - -impl<'a, 'b> Recv<'a, 'b> { - pub(super) fn new(socket: &'a UdpSocket, buf: &'b mut [u8]) -> Self { - Self { socket, buf } - } -} - -impl<'a, 'b> Future for Recv<'a, 'b> { - type Output = io::Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let Recv { socket, buf } = self.get_mut(); - socket.poll_recv_priv(cx, buf) - } -} diff --git a/tokio-udp/src/recv_from.rs b/tokio-udp/src/recv_from.rs deleted file mode 100644 index 04bef680a..000000000 --- a/tokio-udp/src/recv_from.rs +++ /dev/null @@ -1,31 +0,0 @@ -use super::UdpSocket; -use std::future::Future; -use std::io; -use std::net::SocketAddr; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that receives a datagram. -/// -/// This `struct` is created by [`recv_from`](super::UdpSocket::recv_from). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct RecvFrom<'a, 'b> { - socket: &'a UdpSocket, - buf: &'b mut [u8], -} - -impl<'a, 'b> RecvFrom<'a, 'b> { - pub(super) fn new(socket: &'a UdpSocket, buf: &'b mut [u8]) -> Self { - Self { socket, buf } - } -} - -impl<'a, 'b> Future for RecvFrom<'a, 'b> { - type Output = io::Result<(usize, SocketAddr)>; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let RecvFrom { socket, buf } = self.get_mut(); - socket.poll_recv_from_priv(cx, buf) - } -} diff --git a/tokio-udp/src/send.rs b/tokio-udp/src/send.rs deleted file mode 100644 index 8fd1007f4..000000000 --- a/tokio-udp/src/send.rs +++ /dev/null @@ -1,30 +0,0 @@ -use super::UdpSocket; -use std::future::Future; -use std::io; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that sends a datagram to the connected address. -/// -/// This `struct` is created by [`send`](super::UdpSocket::send). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct Send<'a, 'b> { - socket: &'a UdpSocket, - buf: &'b [u8], -} - -impl<'a, 'b> Send<'a, 'b> { - pub(super) fn new(socket: &'a UdpSocket, buf: &'b [u8]) -> Self { - Self { socket, buf } - } -} - -impl<'a, 'b> Future for Send<'a, 'b> { - type Output = io::Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let Send { socket, buf } = self.get_mut(); - socket.poll_send_priv(cx, buf) - } -} diff --git a/tokio-udp/src/send_to.rs b/tokio-udp/src/send_to.rs deleted file mode 100644 index b7f241214..000000000 --- a/tokio-udp/src/send_to.rs +++ /dev/null @@ -1,40 +0,0 @@ -use super::UdpSocket; -use std::future::Future; -use std::io; -use std::net::SocketAddr; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that sends a datagram to a given address. -/// -/// This `struct` is created by [`send_to`](super::UdpSocket::send_to). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct SendTo<'a, 'b> { - socket: &'a UdpSocket, - buf: &'b [u8], - target: &'b SocketAddr, -} - -impl<'a, 'b> SendTo<'a, 'b> { - pub(super) fn new(socket: &'a UdpSocket, buf: &'b [u8], target: &'b SocketAddr) -> Self { - Self { - socket, - buf, - target, - } - } -} - -impl<'a, 'b> Future for SendTo<'a, 'b> { - type Output = io::Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let SendTo { - socket, - buf, - target, - } = self.get_mut(); - socket.poll_send_to_priv(cx, buf, target) - } -} diff --git a/tokio-udp/src/socket.rs b/tokio-udp/src/socket.rs index 5a3c94e89..f55b1d5da 100644 --- a/tokio-udp/src/socket.rs +++ b/tokio-udp/src/socket.rs @@ -1,14 +1,15 @@ use super::split::{split, UdpSocketRecvHalf, UdpSocketSendHalf}; -use super::{Recv, RecvFrom, Send, SendTo}; + +use tokio_reactor::{Handle, PollEvented}; + use futures_core::ready; +use futures_util::future::poll_fn; use mio; use std::convert::TryFrom; use std::fmt; use std::io; use std::net::{self, Ipv4Addr, Ipv6Addr, SocketAddr}; -use std::pin::Pin; use std::task::{Context, Poll}; -use tokio_reactor::{Handle, PollEvented}; /// An I/O object representing a UDP socket. pub struct UdpSocket { @@ -73,30 +74,8 @@ impl UdpSocket { /// will resolve to an error if the socket is not connected. /// /// [`connect`]: #method.connect - pub fn send<'a, 'b>(&'a mut self, buf: &'b [u8]) -> Send<'a, 'b> { - Send::new(self, buf) - } - - /// Sends data on the socket to the remote address to which it is connected. - /// - /// The [`connect`] method will connect this socket to a remote address. This - /// method will fail if the socket is not connected. - /// - /// [`connect`]: #method.connect - /// - /// # Return - /// - /// On success, returns `Poll::Ready(Ok(num_bytes_written))`. - /// - /// If the socket is not ready for writing, the method returns - /// `Poll::Pending` and arranges for the current task to receive a - /// notification when the socket becomes writable. - pub fn poll_send( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - self.poll_send_priv(cx, buf) + pub async fn send(&mut self, buf: &[u8]) -> io::Result { + poll_fn(|cx| self.poll_send_priv(cx, buf)).await } // Poll IO functions that takes `&self` are provided for the split API. @@ -137,35 +116,8 @@ impl UdpSocket { /// will fail if the socket is not connected. /// /// [`connect`]: #method.connect - pub fn recv<'a, 'b>(&'a mut self, buf: &'b mut [u8]) -> Recv<'a, 'b> { - Recv::new(self, buf) - } - - /// Receives a single datagram message on the socket from the remote address to - /// which it is connected. On success, returns the number of bytes read. - /// - /// The function must be called with valid byte array `buf` of sufficient size to - /// hold the message bytes. If a message is too long to fit in the supplied buffer, - /// excess bytes may be discarded. - /// - /// The [`connect`] method will connect this socket to a remote address. This - /// method will fail if the socket is not connected. - /// - /// [`connect`]: #method.connect - /// - /// # Return - /// - /// On success, returns `Poll::Ready(Ok(num_bytes_read))`. - /// - /// If no data is available for reading, the method returns - /// `Poll::Pending` and arranges for the current task to receive a - /// notification when the socket becomes receivable or is closed. - pub fn poll_recv( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { - self.poll_recv_priv(cx, buf) + pub async fn recv(&mut self, buf: &mut [u8]) -> io::Result { + poll_fn(|cx| self.poll_recv_priv(cx, buf)).await } pub(crate) fn poll_recv_priv( @@ -189,30 +141,8 @@ impl UdpSocket { /// /// The future will resolve to an error if the IP version of the socket does /// not match that of `target`. - pub fn send_to<'a, 'b>(&'a mut self, buf: &'b [u8], target: &'b SocketAddr) -> SendTo<'a, 'b> { - SendTo::new(self, buf, target) - } - - /// Sends data on the socket to the given address. On success, returns the - /// number of bytes written. - /// - /// This will return an error when the IP version of the local socket - /// does not match that of `target`. - /// - /// # Return - /// - /// On success, returns `Poll::Ready(Ok(num_bytes_written))`. - /// - /// If the socket is not ready for writing, the method returns - /// `Poll::Pending` and arranges for the current task to receive a - /// notification when the socket becomes writable. - pub fn poll_send_to( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - target: &SocketAddr, - ) -> Poll> { - self.poll_send_to_priv(cx, buf, target) + pub async fn send_to(&mut self, buf: &[u8], target: &SocketAddr) -> io::Result { + poll_fn(|cx| self.poll_send_to_priv(cx, buf, target)).await } pub(crate) fn poll_send_to_priv( @@ -238,18 +168,8 @@ impl UdpSocket { /// The function must be called with valid byte array `buf` of sufficient size /// to hold the message bytes. If a message is too long to fit in the supplied /// buffer, excess bytes may be discarded. - pub fn recv_from<'a, 'b>(&'a mut self, buf: &'b mut [u8]) -> RecvFrom<'a, 'b> { - RecvFrom::new(self, buf) - } - - /// Receives data from the socket. On success, returns the number of bytes - /// read and the address from whence the data came. - pub fn poll_recv_from( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { - self.poll_recv_from_priv(cx, buf) + pub async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + poll_fn(|cx| self.poll_recv_from_priv(cx, buf)).await } pub(crate) fn poll_recv_from_priv( @@ -268,42 +188,6 @@ impl UdpSocket { } } - /// Check the UDP socket's read readiness state. - /// - /// The mask argument allows specifying what readiness to notify on. This - /// can be any value, including platform specific readiness, **except** - /// `writable`. - /// - /// If the socket is not ready for receiving then `Poll::Pending` is - /// returned and the current task is notified once a new event is received. - /// - /// The socket will remain in a read-ready state until calls to `poll_recv` - /// return `Poll::Pending`. - /// - /// # Panics - /// - /// This function panics if: - /// - /// * `ready` includes writable. - pub fn poll_read_ready( - &self, - cx: &mut Context<'_>, - mask: mio::Ready, - ) -> Poll> { - self.io.poll_read_ready(cx, mask) - } - - /// Check the UDP socket's write readiness state. - /// - /// If the socket is not ready for sending then `Poll::Pending` is - /// returned and the current task is notified once a new event is received. - /// - /// The I/O resource will remain in a write-ready state until calls to - /// `poll_send` return `Poll::Pending`. - pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll> { - self.io.poll_write_ready(cx) - } - /// Gets the value of the `SO_BROADCAST` option for this socket. /// /// For more information about this option, see [`set_broadcast`]. diff --git a/tokio-udp/src/split.rs b/tokio-udp/src/split.rs index fc2c6001a..c45bf567e 100644 --- a/tokio-udp/src/split.rs +++ b/tokio-udp/src/split.rs @@ -12,9 +12,12 @@ //! The halves can be reunited to the original socket with their `reunite` //! methods. -use super::{Recv, RecvFrom, Send, SendTo, UdpSocket}; +use super::UdpSocket; + +use futures_util::future::poll_fn; use std::error::Error; use std::fmt; +use std::io; use std::net::SocketAddr; use std::sync::Arc; @@ -82,8 +85,8 @@ impl UdpSocketRecvHalf { /// The function must be called with valid byte array `buf` of sufficient size /// to hold the message bytes. If a message is too long to fit in the supplied /// buffer, excess bytes may be discarded. - pub fn recv_from<'a, 'b>(&'a mut self, buf: &'b mut [u8]) -> RecvFrom<'a, 'b> { - RecvFrom::new(&self.0, buf) + pub async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + poll_fn(|cx| self.0.poll_recv_from_priv(cx, buf)).await } /// Returns a future that receives a single datagram message on the socket from @@ -98,8 +101,8 @@ impl UdpSocketRecvHalf { /// will fail if the socket is not connected. /// /// [`connect`]: super::UdpSocket::connect - pub fn recv<'a, 'b>(&'a mut self, buf: &'b mut [u8]) -> Recv<'a, 'b> { - Recv::new(&self.0, buf) + pub async fn recv<'a, 'b>(&'a mut self, buf: &'b mut [u8]) -> io::Result { + poll_fn(|cx| self.0.poll_recv_priv(cx, buf)).await } } @@ -116,8 +119,8 @@ impl UdpSocketSendHalf { /// /// The future will resolve to an error if the IP version of the socket does /// not match that of `target`. - pub fn send_to<'a, 'b>(&'a mut self, buf: &'b [u8], target: &'b SocketAddr) -> SendTo<'a, 'b> { - SendTo::new(&self.0, buf, target) + pub async fn send_to(&mut self, buf: &[u8], target: &SocketAddr) -> io::Result { + poll_fn(|cx| self.0.poll_send_to_priv(cx, buf, target)).await } /// Returns a future that sends data on the socket to the remote address to which it is connected. @@ -127,8 +130,8 @@ impl UdpSocketSendHalf { /// will resolve to an error if the socket is not connected. /// /// [`connect`]: super::UdpSocket::connect - pub fn send<'a, 'b>(&'a mut self, buf: &'b [u8]) -> Send<'a, 'b> { - Send::new(&self.0, buf) + pub async fn send(&mut self, buf: &[u8]) -> io::Result { + poll_fn(|cx| self.0.poll_send_priv(cx, buf)).await } } diff --git a/tokio/src/net.rs b/tokio/src/net.rs index 131071d92..ba5157e3a 100644 --- a/tokio/src/net.rs +++ b/tokio/src/net.rs @@ -58,7 +58,7 @@ pub mod udp { //! [`Send`]: struct.Send.html //! [`RecvFrom`]: struct.RecvFrom.html //! [`SendTo`]: struct.SendTo.html - pub use tokio_udp::{split, Recv, RecvFrom, Send, SendTo, UdpSocket}; + pub use tokio_udp::{split, UdpSocket}; } #[cfg(feature = "udp")] pub use self::udp::UdpSocket;