tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-17 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

io: driver internal cleanup (#3124)

Browse Source 
* Removes duplicated code by moving it to `Registration`.
* impl `Deref` for `PollEvented` to avoid `get_ref()`.
* Avoid extra waker clones in I/O driver.
* Add `Interest` wrapper around `mio::Interest`.
This commit is contained in:
Carl Lerche
2020-11-11 09:28:21 -08:00
committed by GitHub
parent d869e16990
commit ce891a4df1
18 changed files with 440 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/io/async_fd.rs
+7 -8
View File
@@ -1,5 +1,4 @@
use crate::io::driver::{Direction, Handle, ReadyEvent};
use crate::io::registration::Registration;
use crate::io::driver::{Handle, Interest, ReadyEvent, Registration};
use mio::unix::SourceFd;
use std::io;
@@ -74,7 +73,7 @@ pub struct AsyncFdReadyGuard<'a, T: AsRawFd> {
event: Option<ReadyEvent>,
}
const ALL_INTEREST: mio::Interest = mio::Interest::READABLE.add(mio::Interest::WRITABLE);
const ALL_INTEREST: Interest = Interest::READABLE.add(Interest::WRITABLE);
impl<T: AsRawFd> AsyncFd<T> {
/// Creates an AsyncFd backed by (and taking ownership of) an object
@@ -145,7 +144,7 @@ impl<T: AsRawFd> AsyncFd<T> {
&'a self,
cx: &mut Context<'_>,
) -> Poll<io::Result<AsyncFdReadyGuard<'a, T>>> {
let event = ready!(self.registration.poll_readiness(cx, Direction::Read))?;
let event = ready!(self.registration.poll_read_ready(cx))?;
Ok(AsyncFdReadyGuard {
async_fd: self,
@@ -170,7 +169,7 @@ impl<T: AsRawFd> AsyncFd<T> {
&'a self,
cx: &mut Context<'_>,
) -> Poll<io::Result<AsyncFdReadyGuard<'a, T>>> {
let event = ready!(self.registration.poll_readiness(cx, Direction::Write))?;
let event = ready!(self.registration.poll_write_ready(cx))?;
Ok(AsyncFdReadyGuard {
async_fd: self,
@@ -179,7 +178,7 @@ impl<T: AsRawFd> AsyncFd<T> {
.into()
}
async fn readiness(&self, interest: mio::Interest) -> io::Result<AsyncFdReadyGuard<'_, T>> {
async fn readiness(&self, interest: Interest) -> io::Result<AsyncFdReadyGuard<'_, T>> {
let event = self.registration.readiness(interest).await?;
Ok(AsyncFdReadyGuard {
@@ -193,7 +192,7 @@ impl<T: AsRawFd> AsyncFd<T> {
///
/// [`AsyncFdReadyGuard`]: struct@self::AsyncFdReadyGuard
pub async fn readable(&self) -> io::Result<AsyncFdReadyGuard<'_, T>> {
self.readiness(mio::Interest::READABLE).await
self.readiness(Interest::READABLE).await
}
/// Waits for the file descriptor to become writable, returning a
@@ -201,7 +200,7 @@ impl<T: AsRawFd> AsyncFd<T> {
///
/// [`AsyncFdReadyGuard`]: struct@self::AsyncFdReadyGuard
pub async fn writable(&self) -> io::Result<AsyncFdReadyGuard<'_, T>> {
self.readiness(mio::Interest::WRITABLE).await
self.readiness(Interest::WRITABLE).await
}
}
tokio/src/io/driver/interest.rs
+58
View File
@@ -0,0 +1,58 @@
use std::fmt;
use std::ops;
/// Readiness event interest
///
/// Specifies the readiness events the caller is interested in when awaiting on
/// I/O resource readiness states.
#[derive(Clone, Copy)]
pub(crate) struct Interest(mio::Interest);
impl Interest {
/// Interest in all readable events
pub(crate) const READABLE: Interest = Interest(mio::Interest::READABLE);
/// Interest in all writable events
pub(crate) const WRITABLE: Interest = Interest(mio::Interest::WRITABLE);
/// Returns true if the value includes readable interest.
pub(crate) const fn is_readable(self) -> bool {
self.0.is_readable()
}
/// Returns true if the value includes writable interest.
pub(crate) const fn is_writable(self) -> bool {
self.0.is_writable()
}
/// Add together two `Interst` values.
pub(crate) const fn add(self, other: Interest) -> Interest {
Interest(self.0.add(other.0))
}
pub(crate) const fn to_mio(self) -> mio::Interest {
self.0
}
}
impl ops::BitOr for Interest {
type Output = Self;
#[inline]
fn bitor(self, other: Self) -> Self {
self.add(other)
}
}
impl ops::BitOrAssign for Interest {
#[inline]
fn bitor_assign(&mut self, other: Self) {
self.0 = (*self | other).0;
}
}
impl fmt::Debug for Interest {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(fmt)
}
}
tokio/src/io/driver/mod.rs
+10 -4
View File
@@ -1,10 +1,16 @@
#![cfg_attr(not(feature = "rt"), allow(dead_code))]
mod interest;
pub(crate) use interest::Interest;
mod ready;
use ready::Ready;
mod registration;
pub(crate) use registration::Registration;
mod scheduled_io;
pub(crate) use scheduled_io::ScheduledIo; // pub(crate) for tests
use scheduled_io::ScheduledIo;
use crate::park::{Park, Unpark};
use crate::util::slab::{self, Slab};
@@ -68,7 +74,7 @@ pub(super) struct Inner {
}
#[derive(Debug, Eq, PartialEq, Clone, Copy)]
pub(super) enum Direction {
enum Direction {
Read,
Write,
}
@@ -313,7 +319,7 @@ impl Inner {
pub(super) fn add_source(
&self,
source: &mut impl mio::event::Source,
interest: mio::Interest,
interest: Interest,
) -> io::Result<slab::Ref<ScheduledIo>> {
let (address, shared) = self.io_dispatch.allocate().ok_or_else(|| {
io::Error::new(
@@ -325,7 +331,7 @@ impl Inner {
let token = GENERATION.pack(shared.generation(), ADDRESS.pack(address.as_usize(), 0));
self.registry
.register(source, mio::Token(token), interest)?;
.register(source, mio::Token(token), interest.to_mio())?;
Ok(shared)
}
tokio/src/io/driver/ready.rs
+5 -3
View File
@@ -114,8 +114,10 @@ impl Ready {
}
cfg_io_readiness! {
use crate::io::Interest;
impl Ready {
pub(crate) fn from_interest(interest: mio::Interest) -> Ready {
pub(crate) fn from_interest(interest: Interest) -> Ready {
let mut ready = Ready::EMPTY;
if interest.is_readable() {
@@ -131,11 +133,11 @@ cfg_io_readiness! {
ready
}
pub(crate) fn intersection(self, interest: mio::Interest) -> Ready {
pub(crate) fn intersection(self, interest: Interest) -> Ready {
Ready(self.0 & Ready::from_interest(interest).0)
}
pub(crate) fn satisfies(self, interest: mio::Interest) -> bool {
pub(crate) fn satisfies(self, interest: Interest) -> bool {
self.0 & Ready::from_interest(interest).0 != 0
}
}
tokio/src/io/registration.rs → tokio/src/io/driver/registration.rs
+78 -11
View File
@@ -1,4 +1,4 @@
use crate::io::driver::{Direction, Handle, ReadyEvent, ScheduledIo};
use crate::io::driver::{Direction, Handle, Interest, ReadyEvent, ScheduledIo};
use crate::util::slab;
use mio::event::Source;
@@ -53,11 +53,12 @@ unsafe impl Sync for Registration {}
// ===== impl Registration =====
impl Registration {
/// Registers the I/O resource with the default reactor, for a specific `mio::Interest`.
/// `new_with_interest` should be used over `new` when you need control over the readiness state,
/// such as when a file descriptor only allows reads. This does not add `hup` or `error` so if
/// you are interested in those states, you will need to add them to the readiness state passed
/// to this function.
/// Registers the I/O resource with the default reactor, for a specific
/// `Interest`. `new_with_interest` should be used over `new` when you need
/// control over the readiness state, such as when a file descriptor only
/// allows reads. This does not add `hup` or `error` so if you are
/// interested in those states, you will need to add them to the readiness
/// state passed to this function.
///
/// # Return
///
@@ -65,7 +66,7 @@ impl Registration {
/// - `Err` if an error was encountered during registration
pub(crate) fn new_with_interest_and_handle(
io: &mut impl Source,
interest: mio::Interest,
interest: Interest,
handle: Handle,
) -> io::Result<Registration> {
let shared = if let Some(inner) = handle.inner() {
@@ -96,7 +97,7 @@ impl Registration {
/// no longer result in notifications getting sent for this registration.
///
/// `Err` is returned if an error is encountered.
pub(super) fn deregister(&mut self, io: &mut impl Source) -> io::Result<()> {
pub(crate) fn deregister(&mut self, io: &mut impl Source) -> io::Result<()> {
let inner = match self.handle.inner() {
Some(inner) => inner,
None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
@@ -104,15 +105,47 @@ impl Registration {
inner.deregister_source(io)
}
pub(super) fn clear_readiness(&self, event: ReadyEvent) {
pub(crate) fn clear_readiness(&self, event: ReadyEvent) {
self.shared.clear_readiness(event);
}
// Uses the poll path, requiring the caller to ensure mutual exclusion for
// correctness. Only the last task to call this function is notified.
pub(crate) fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<ReadyEvent>> {
self.poll_ready(cx, Direction::Read)
}
// Uses the poll path, requiring the caller to ensure mutual exclusion for
// correctness. Only the last task to call this function is notified.
pub(crate) fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<ReadyEvent>> {
self.poll_ready(cx, Direction::Write)
}
// Uses the poll path, requiring the caller to ensure mutual exclusion for
// correctness. Only the last task to call this function is notified.
pub(crate) fn poll_read_io<R>(
&self,
cx: &mut Context<'_>,
f: impl FnMut() -> io::Result<R>,
) -> Poll<io::Result<R>> {
self.poll_io(cx, Direction::Read, f)
}
// Uses the poll path, requiring the caller to ensure mutual exclusion for
// correctness. Only the last task to call this function is notified.
pub(crate) fn poll_write_io<R>(
&self,
cx: &mut Context<'_>,
f: impl FnMut() -> io::Result<R>,
) -> Poll<io::Result<R>> {
self.poll_io(cx, Direction::Write, f)
}
/// Polls for events on the I/O resource's `direction` readiness stream.
///
/// If called with a task context, notify the task when a new event is
/// received.
pub(super) fn poll_readiness(
fn poll_ready(
&self,
cx: &mut Context<'_>,
direction: Direction,
@@ -128,6 +161,27 @@ impl Registration {
coop.made_progress();
Poll::Ready(Ok(ev))
}
fn poll_io<R>(
&self,
cx: &mut Context<'_>,
direction: Direction,
mut f: impl FnMut() -> io::Result<R>,
) -> Poll<io::Result<R>> {
loop {
let ev = ready!(self.poll_ready(cx, direction))?;
match f() {
Ok(ret) => {
return Poll::Ready(Ok(ret));
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.clear_readiness(ev);
}
Err(e) => return Poll::Ready(Err(e)),
}
}
}
}
fn gone() -> io::Error {
@@ -136,7 +190,7 @@ fn gone() -> io::Error {
cfg_io_readiness! {
impl Registration {
pub(super) async fn readiness(&self, interest: mio::Interest) -> io::Result<ReadyEvent> {
pub(crate) async fn readiness(&self, interest: Interest) -> io::Result<ReadyEvent> {
use std::future::Future;
use std::pin::Pin;
@@ -151,5 +205,18 @@ cfg_io_readiness! {
Pin::new(&mut fut).poll(cx).map(Ok)
}).await
}
pub(crate) async fn async_io<R>(&self, interest: Interest, mut f: impl FnMut() -> io::Result<R>) -> io::Result<R> {
loop {
let event = self.readiness(interest).await?;
match f() {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.clear_readiness(event);
}
x => return x,
}
}
}
}
}
tokio/src/io/driver/scheduled_io.rs
+19 -5
View File
@@ -49,6 +49,8 @@ struct Waiters {
}
cfg_io_readiness! {
use crate::io::Interest;
#[derive(Debug)]
struct Waiter {
pointers: linked_list::Pointers<Waiter>,
@@ -57,7 +59,7 @@ cfg_io_readiness! {
waker: Option<Waker>,
/// The interest this waiter is waiting on
interest: mio::Interest,
interest: Interest,
is_ready: bool,
@@ -283,7 +285,7 @@ impl ScheduledIo {
/// These are to support `AsyncRead` and `AsyncWrite` polling methods,
/// which cannot use the `async fn` version. This uses reserved reader
/// and writer slots.
pub(in crate::io) fn poll_readiness(
pub(super) fn poll_readiness(
&self,
cx: &mut Context<'_>,
direction: Direction,
@@ -299,7 +301,19 @@ impl ScheduledIo {
Direction::Read => &mut waiters.reader,
Direction::Write => &mut waiters.writer,
};
*slot = Some(cx.waker().clone());
// Avoid cloning the waker if one is already stored that matches the
// current task.
match slot {
Some(existing) => {
if !existing.will_wake(cx.waker()) {
*existing = cx.waker().clone();
}
}
None => {
*slot = Some(cx.waker().clone());
}
}
// Try again, in case the readiness was changed while we were
// taking the waiters lock
@@ -348,7 +362,7 @@ unsafe impl Sync for ScheduledIo {}
cfg_io_readiness! {
impl ScheduledIo {
/// An async version of `poll_readiness` which uses a linked list of wakers
pub(crate) async fn readiness(&self, interest: mio::Interest) -> ReadyEvent {
pub(crate) async fn readiness(&self, interest: Interest) -> ReadyEvent {
self.readiness_fut(interest).await
}
@@ -356,7 +370,7 @@ cfg_io_readiness! {
// we are borrowing the `UnsafeCell` possibly over await boundaries.
//
// Go figure.
fn readiness_fut(&self, interest: mio::Interest) -> Readiness<'_> {
fn readiness_fut(&self, interest: Interest) -> Readiness<'_> {
Readiness {
scheduled_io: self,
state: State::Init,
tokio/src/io/mod.rs
+1 -2
View File
@@ -206,8 +206,7 @@ pub use std::io::{Error, ErrorKind, Result, SeekFrom};
cfg_io_driver! {
pub(crate) mod driver;
mod registration;
pub(crate) use driver::Interest;
mod poll_evented;
tokio/src/io/poll_evented.rs
+55 -177
View File
@@ -1,13 +1,9 @@
use crate::io::driver::{Direction, Handle, ReadyEvent};
use crate::io::registration::Registration;
use crate::io::{AsyncRead, AsyncWrite, ReadBuf};
use crate::io::driver::{Handle, Interest, Registration};
use mio::event::Source;
use std::fmt;
use std::io::{self, Read, Write};
use std::marker::Unpin;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::io;
use std::ops::Deref;
cfg_io_driver! {
/// Associates an I/O resource that implements the [`std::io::Read`] and/or
@@ -89,30 +85,32 @@ impl<E: Source> PollEvented<E> {
/// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
#[cfg_attr(feature = "signal", allow(unused))]
pub(crate) fn new(io: E) -> io::Result<Self> {
PollEvented::new_with_interest(io, mio::Interest::READABLE | mio::Interest::WRITABLE)
PollEvented::new_with_interest(io, Interest::READABLE | Interest::WRITABLE)
}
/// Creates a new `PollEvented` associated with the default reactor, for specific `mio::Interest`
/// state. `new_with_interest` should be used over `new` when you need control over the readiness
/// state, such as when a file descriptor only allows reads. This does not add `hup` or `error`
/// so if you are interested in those states, you will need to add them to the readiness state
/// passed to this function.
/// Creates a new `PollEvented` associated with the default reactor, for
/// specific `Interest` state. `new_with_interest` should be used over `new`
/// when you need control over the readiness state, such as when a file
/// descriptor only allows reads. This does not add `hup` or `error` so if
/// you are interested in those states, you will need to add them to the
/// readiness state passed to this function.
///
/// # Panics
///
/// This function panics if thread-local runtime is not set.
///
/// The runtime is usually set implicitly when this function is called
/// from a future driven by a tokio runtime, otherwise runtime can be set
/// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
/// The runtime is usually set implicitly when this function is called from
/// a future driven by a tokio runtime, otherwise runtime can be set
/// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter)
/// function.
#[cfg_attr(feature = "signal", allow(unused))]
pub(crate) fn new_with_interest(io: E, interest: mio::Interest) -> io::Result<Self> {
pub(crate) fn new_with_interest(io: E, interest: Interest) -> io::Result<Self> {
Self::new_with_interest_and_handle(io, interest, Handle::current())
}
pub(crate) fn new_with_interest_and_handle(
mut io: E,
interest: mio::Interest,
interest: Interest,
handle: Handle,
) -> io::Result<Self> {
let registration = Registration::new_with_interest_and_handle(&mut io, interest, handle)?;
@@ -122,177 +120,57 @@ impl<E: Source> PollEvented<E> {
})
}
/// Returns a shared reference to the underlying I/O object this readiness
/// stream is wrapping.
#[cfg(any(feature = "net", feature = "process", feature = "signal"))]
pub(crate) fn get_ref(&self) -> &E {
self.io.as_ref().unwrap()
}
/// Returns a mutable reference to the underlying I/O object this readiness
/// stream is wrapping.
pub(crate) fn get_mut(&mut self) -> &mut E {
self.io.as_mut().unwrap()
}
pub(crate) fn clear_readiness(&self, event: ReadyEvent) {
self.registration.clear_readiness(event);
}
/// Checks the I/O resource'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`. HUP is always implicitly included on platforms that support
/// it.
///
/// If the resource is not ready for a read 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 read-ready state until readiness is
/// cleared by calling [`clear_read_ready`].
///
/// [`clear_read_ready`]: method@Self::clear_read_ready
///
/// # Panics
///
/// This function panics if:
///
/// * `ready` includes writable.
/// * called from outside of a task context.
///
/// # Warning
///
/// This method may not be called concurrently. It takes `&self` to allow
/// calling it concurrently with `poll_write_ready`.
pub(crate) fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<ReadyEvent>> {
self.registration.poll_readiness(cx, Direction::Read)
}
/// Checks the I/O resource's write readiness state.
///
/// This always checks for writable readiness and also checks for HUP
/// readiness on platforms that support it.
///
/// If the resource is not ready for a write 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 readiness is
/// cleared by calling [`clear_write_ready`].
///
/// [`clear_write_ready`]: method@Self::clear_write_ready
///
/// # Panics
///
/// This function panics if:
///
/// * `ready` contains bits besides `writable` and `hup`.
/// * called from outside of a task context.
///
/// # Warning
///
/// This method may not be called concurrently. It takes `&self` to allow
/// calling it concurrently with `poll_read_ready`.
pub(crate) fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<ReadyEvent>> {
self.registration.poll_readiness(cx, Direction::Write)
/// Returns a reference to the registration
pub(crate) fn registration(&self) -> &Registration {
&self.registration
}
}
cfg_io_readiness! {
feature! {
#![any(feature = "net", feature = "process")]
use crate::io::ReadBuf;
use std::task::{Context, Poll};
impl<E: Source> PollEvented<E> {
pub(crate) async fn readiness(&self, interest: mio::Interest) -> io::Result<ReadyEvent> {
self.registration.readiness(interest).await
}
pub(crate) async fn async_io<F, R>(&self, interest: mio::Interest, mut op: F) -> io::Result<R>
// Safety: The caller must ensure that `E` can read into uninitialized memory
pub(crate) unsafe fn poll_read<'a>(
&'a self,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>>
where
F: FnMut(&E) -> io::Result<R>,
&'a E: io::Read + 'a,
{
loop {
let event = self.readiness(interest).await?;
use std::io::Read;
match op(self.get_ref()) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.clear_readiness(event);
}
x => return x,
}
}
let n = ready!(self.registration.poll_read_io(cx, || {
let b = &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]);
self.io.as_ref().unwrap().read(b)
}))?;
// Safety: We trust `TcpStream::read` to have filled up `n` bytes in the
// buffer.
buf.assume_init(n);
buf.advance(n);
Poll::Ready(Ok(()))
}
pub(crate) fn poll_write<'a>(&'a self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>>
where
&'a E: io::Write + 'a,
{
use std::io::Write;
self.registration.poll_write_io(cx, || self.io.as_ref().unwrap().write(buf))
}
}
}
// ===== Read / Write impls =====
impl<E: Source> Deref for PollEvented<E> {
type Target = E;
impl<E: Source + Read + Unpin> AsyncRead for PollEvented<E> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
loop {
let ev = ready!(self.poll_read_ready(cx))?;
// We can't assume the `Read` won't look at the read buffer,
// so we have to force initialization here.
let r = (*self).get_mut().read(buf.initialize_unfilled());
if is_wouldblock(&r) {
self.clear_readiness(ev);
continue;
}
return Poll::Ready(r.map(|n| {
buf.advance(n);
}));
}
}
}
impl<E: Source + Write + Unpin> AsyncWrite for PollEvented<E> {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
loop {
let ev = ready!(self.poll_write_ready(cx))?;
let r = (*self).get_mut().write(buf);
if is_wouldblock(&r) {
self.clear_readiness(ev);
continue;
}
return Poll::Ready(r);
}
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
loop {
let ev = ready!(self.poll_write_ready(cx))?;
let r = (*self).get_mut().flush();
if is_wouldblock(&r) {
self.clear_readiness(ev);
continue;
}
return Poll::Ready(r);
}
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
match *r {
Ok(_) => false,
Err(ref e) => e.kind() == io::ErrorKind::WouldBlock,
fn deref(&self) -> &E {
self.io.as_ref().unwrap()
}
}
tokio/src/macros/cfg.rs
+13
View File
@@ -1,5 +1,18 @@
#![allow(unused_macros)]
macro_rules! feature {
(
#![$meta:meta]
$($item:item)*
) => {
$(
#[cfg($meta)]
#[cfg_attr(docsrs, doc(cfg($meta)))]
$item
)*
}
}
/// Enables enter::block_on
macro_rules! cfg_block_on {
($($item:item)*) => {
tokio/src/net/tcp/listener.rs
+12 -11
View File
@@ -1,4 +1,4 @@
use crate::io::PollEvented;
use crate::io::{Interest, PollEvented};
use crate::net::tcp::TcpStream;
use crate::net::{to_socket_addrs, ToSocketAddrs};
@@ -164,7 +164,8 @@ impl TcpListener {
pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
let (mio, addr) = self
.io
.async_io(mio::Interest::READABLE, |sock| sock.accept())
.registration()
.async_io(Interest::READABLE, || self.io.accept())
.await?;
let stream = TcpStream::new(mio)?;
@@ -181,15 +182,15 @@ impl TcpListener {
/// single task. Failing to do this could result in tasks hanging.
pub fn poll_accept(&self, cx: &mut Context<'_>) -> Poll<io::Result<(TcpStream, SocketAddr)>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
let ev = ready!(self.io.registration().poll_read_ready(cx))?;
match self.io.get_ref().accept() {
match self.io.accept() {
Ok((io, addr)) => {
let io = TcpStream::new(io)?;
return Poll::Ready(Ok((io, addr)));
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
self.io.registration().clear_readiness(ev);
}
Err(e) => return Poll::Ready(Err(e)),
}
@@ -266,7 +267,7 @@ impl TcpListener {
/// }
/// ```
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().local_addr()
self.io.local_addr()
}
/// Gets the value of the `IP_TTL` option for this socket.
@@ -293,7 +294,7 @@ impl TcpListener {
/// }
/// ```
pub fn ttl(&self) -> io::Result<u32> {
self.io.get_ref().ttl()
self.io.ttl()
}
/// Sets the value for the `IP_TTL` option on this socket.
@@ -318,7 +319,7 @@ impl TcpListener {
/// }
/// ```
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
self.io.get_ref().set_ttl(ttl)
self.io.set_ttl(ttl)
}
}
@@ -346,7 +347,7 @@ impl TryFrom<net::TcpListener> for TcpListener {
impl fmt::Debug for TcpListener {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.io.get_ref().fmt(f)
self.io.fmt(f)
}
}
@@ -357,7 +358,7 @@ mod sys {
impl AsRawFd for TcpListener {
fn as_raw_fd(&self) -> RawFd {
self.io.get_ref().as_raw_fd()
self.io.as_raw_fd()
}
}
}
@@ -369,7 +370,7 @@ mod sys {
impl AsRawSocket for TcpListener {
fn as_raw_socket(&self) -> RawSocket {
self.io.get_ref().as_raw_socket()
self.io.as_raw_socket()
}
}
}
tokio/src/net/tcp/stream.rs
+22 -51
View File
@@ -1,12 +1,12 @@
use crate::future::poll_fn;
use crate::io::{AsyncRead, AsyncWrite, PollEvented, ReadBuf};
use crate::io::{AsyncRead, AsyncWrite, Interest, PollEvented, ReadBuf};
use crate::net::tcp::split::{split, ReadHalf, WriteHalf};
use crate::net::tcp::split_owned::{split_owned, OwnedReadHalf, OwnedWriteHalf};
use crate::net::{to_socket_addrs, ToSocketAddrs};
use std::convert::TryFrom;
use std::fmt;
use std::io::{self, Read, Write};
use std::io;
use std::net::{Shutdown, SocketAddr};
use std::pin::Pin;
use std::task::{Context, Poll};
@@ -129,9 +129,9 @@ impl TcpStream {
// actually hit an error or not.
//
// If all that succeeded then we ship everything on up.
poll_fn(|cx| stream.io.poll_write_ready(cx)).await?;
poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;
if let Some(e) = stream.io.get_ref().take_error()? {
if let Some(e) = stream.io.take_error()? {
return Err(e);
}
@@ -193,7 +193,7 @@ impl TcpStream {
/// # }
/// ```
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().local_addr()
self.io.local_addr()
}
/// Returns the remote address that this stream is connected to.
@@ -211,7 +211,7 @@ impl TcpStream {
/// # }
/// ```
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().peer_addr()
self.io.peer_addr()
}
/// Attempts to receive data on the socket, without removing that data from
@@ -252,12 +252,12 @@ impl TcpStream {
/// ```
pub fn poll_peek(&self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
let ev = ready!(self.io.registration().poll_read_ready(cx))?;
match self.io.get_ref().peek(buf) {
match self.io.peek(buf) {
Ok(ret) => return Poll::Ready(Ok(ret)),
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
self.io.registration().clear_readiness(ev);
}
Err(e) => return Poll::Ready(Err(e)),
}
@@ -303,7 +303,8 @@ impl TcpStream {
/// [`AsyncReadExt`]: trait@crate::io::AsyncReadExt
pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
self.io
.async_io(mio::Interest::READABLE, |io| io.peek(buf))
.registration()
.async_io(Interest::READABLE, || self.io.peek(buf))
.await
}
@@ -332,7 +333,7 @@ impl TcpStream {
/// }
/// ```
pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
self.io.get_ref().shutdown(how)
self.io.shutdown(how)
}
/// Gets the value of the `TCP_NODELAY` option on this socket.
@@ -354,7 +355,7 @@ impl TcpStream {
/// # }
/// ```
pub fn nodelay(&self) -> io::Result<bool> {
self.io.get_ref().nodelay()
self.io.nodelay()
}
/// Sets the value of the `TCP_NODELAY` option on this socket.
@@ -378,7 +379,7 @@ impl TcpStream {
/// # }
/// ```
pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
self.io.get_ref().set_nodelay(nodelay)
self.io.set_nodelay(nodelay)
}
/// Gets the value of the `IP_TTL` option for this socket.
@@ -400,7 +401,7 @@ impl TcpStream {
/// # }
/// ```
pub fn ttl(&self) -> io::Result<u32> {
self.io.get_ref().ttl()
self.io.ttl()
}
/// Sets the value for the `IP_TTL` option on this socket.
@@ -421,7 +422,7 @@ impl TcpStream {
/// # }
/// ```
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
self.io.get_ref().set_ttl(ttl)
self.io.set_ttl(ttl)
}
// These lifetime markers also appear in the generated documentation, and make
@@ -469,29 +470,8 @@ impl TcpStream {
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
// Safety: `TcpStream::read` will not peek at the maybe uinitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
match self.io.get_ref().read(b) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
Ok(n) => {
// Safety: We trust `TcpStream::read` to have filled up `n` bytes
// in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
return Poll::Ready(Ok(()));
}
Err(e) => return Poll::Ready(Err(e)),
}
}
// Safety: `TcpStream::read` correctly handles reads into uninitialized memory
unsafe { self.io.poll_read(cx, buf) }
}
pub(super) fn poll_write_priv(
@@ -499,16 +479,7 @@ impl TcpStream {
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
loop {
let ev = ready!(self.io.poll_write_ready(cx))?;
match self.io.get_ref().write(buf) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
x => return Poll::Ready(x),
}
}
self.io.poll_write(cx, buf)
}
}
@@ -559,7 +530,7 @@ impl AsyncWrite for TcpStream {
impl fmt::Debug for TcpStream {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.io.get_ref().fmt(f)
self.io.fmt(f)
}
}
@@ -570,7 +541,7 @@ mod sys {
impl AsRawFd for TcpStream {
fn as_raw_fd(&self) -> RawFd {
self.io.get_ref().as_raw_fd()
self.io.as_raw_fd()
}
}
}
@@ -582,7 +553,7 @@ mod sys {
impl AsRawSocket for TcpStream {
fn as_raw_socket(&self) -> RawSocket {
self.io.get_ref().as_raw_socket()
self.io.as_raw_socket()
}
}
}
tokio/src/net/udp/socket.rs
+69 -102
View File
@@ -1,4 +1,4 @@
use crate::io::{PollEvented, ReadBuf};
use crate::io::{Interest, PollEvented, ReadBuf};
use crate::net::{to_socket_addrs, ToSocketAddrs};
use std::convert::TryFrom;
@@ -216,7 +216,7 @@ impl UdpSocket {
/// # }
/// ```
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().local_addr()
self.io.local_addr()
}
/// Connects the UDP socket setting the default destination for send() and
@@ -248,7 +248,7 @@ impl UdpSocket {
let mut last_err = None;
for addr in addrs {
match self.io.get_ref().connect(addr) {
match self.io.connect(addr) {
Ok(_) => return Ok(()),
Err(e) => last_err = Some(e),
}
@@ -271,7 +271,8 @@ impl UdpSocket {
/// [`connect`]: method@Self::connect
pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
self.io
.async_io(mio::Interest::WRITABLE, |sock| sock.send(buf))
.registration()
.async_io(Interest::WRITABLE, || self.io.send(buf))
.await
}
@@ -299,16 +300,9 @@ impl UdpSocket {
///
/// [`connect`]: method@Self::connect
pub fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
loop {
let ev = ready!(self.io.poll_write_ready(cx))?;
match self.io.get_ref().send(buf) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
x => return Poll::Ready(x),
}
}
self.io
.registration()
.poll_write_io(cx, || self.io.send(buf))
}
/// Try to send data on the socket to the remote address to which it is
@@ -322,7 +316,7 @@ impl UdpSocket {
///
/// [`ErrorKind::WouldBlock`]: std::io::ErrorKind::WouldBlock
pub fn try_send(&self, buf: &[u8]) -> io::Result<usize> {
self.io.get_ref().send(buf)
self.io.send(buf)
}
/// Returns a future that receives a single datagram message on the socket from
@@ -339,7 +333,8 @@ impl UdpSocket {
/// [`connect`]: method@Self::connect
pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
self.io
.async_io(mio::Interest::READABLE, |sock| sock.recv(buf))
.registration()
.async_io(Interest::READABLE, || self.io.recv(buf))
.await
}
@@ -367,29 +362,21 @@ impl UdpSocket {
///
/// [`connect`]: method@Self::connect
pub fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
let n = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
match self.io.get_ref().recv(b) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
Err(e) => return Poll::Ready(Err(e)),
Ok(n) => {
// Safety: We trust `recv` to have filled up `n` bytes
// in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
return Poll::Ready(Ok(()));
}
}
self.io.recv(b)
}))?;
// Safety: We trust `recv` to have filled up `n` bytes in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
Poll::Ready(Ok(()))
}
/// Returns a future that sends data on the socket to the given address.
@@ -448,16 +435,9 @@ impl UdpSocket {
buf: &[u8],
target: &SocketAddr,
) -> Poll<io::Result<usize>> {
loop {
let ev = ready!(self.io.poll_write_ready(cx))?;
match self.io.get_ref().send_to(buf, *target) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
x => return Poll::Ready(x),
}
}
self.io
.registration()
.poll_write_io(cx, || self.io.send_to(buf, *target))
}
/// Try to send data on the socket to the given address, but if the send is blocked
@@ -489,12 +469,13 @@ impl UdpSocket {
///
/// [`ErrorKind::WouldBlock`]: std::io::ErrorKind::WouldBlock
pub fn try_send_to(&self, buf: &[u8], target: SocketAddr) -> io::Result<usize> {
self.io.get_ref().send_to(buf, target)
self.io.send_to(buf, target)
}
async fn send_to_addr(&self, buf: &[u8], target: SocketAddr) -> io::Result<usize> {
self.io
.async_io(mio::Interest::WRITABLE, |sock| sock.send_to(buf, target))
.registration()
.async_io(Interest::WRITABLE, || self.io.send_to(buf, target))
.await
}
@@ -522,7 +503,8 @@ impl UdpSocket {
/// ```
pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
self.io
.async_io(mio::Interest::READABLE, |sock| sock.recv_from(buf))
.registration()
.async_io(Interest::READABLE, || self.io.recv_from(buf))
.await
}
@@ -548,29 +530,21 @@ impl UdpSocket {
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<SocketAddr>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
let (n, addr) = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
match self.io.get_ref().recv_from(b) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
Err(e) => return Poll::Ready(Err(e)),
Ok((n, addr)) => {
// Safety: We trust `recv` to have filled up `n` bytes
// in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
return Poll::Ready(Ok(addr));
}
}
self.io.recv_from(b)
}))?;
// Safety: We trust `recv` to have filled up `n` bytes in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
Poll::Ready(Ok(addr))
}
/// Receives data from the socket, without removing it from the input queue.
@@ -602,7 +576,8 @@ impl UdpSocket {
/// ```
pub async fn peek_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
self.io
.async_io(mio::Interest::READABLE, |sock| sock.peek_from(buf))
.registration()
.async_io(Interest::READABLE, || self.io.peek_from(buf))
.await
}
@@ -637,29 +612,21 @@ impl UdpSocket {
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<SocketAddr>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
let (n, addr) = ready!(self.io.registration().poll_read_io(cx, || {
// Safety: will not read the maybe uinitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
match self.io.get_ref().peek_from(b) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
Err(e) => return Poll::Ready(Err(e)),
Ok((n, addr)) => {
// Safety: We trust `recv` to have filled up `n` bytes
// in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
return Poll::Ready(Ok(addr));
}
}
self.io.peek_from(b)
}))?;
// Safety: We trust `recv` to have filled up `n` bytes in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
Poll::Ready(Ok(addr))
}
/// Gets the value of the `SO_BROADCAST` option for this socket.
@@ -668,7 +635,7 @@ impl UdpSocket {
///
/// [`set_broadcast`]: method@Self::set_broadcast
pub fn broadcast(&self) -> io::Result<bool> {
self.io.get_ref().broadcast()
self.io.broadcast()
}
/// Sets the value of the `SO_BROADCAST` option for this socket.
@@ -676,7 +643,7 @@ impl UdpSocket {
/// When enabled, this socket is allowed to send packets to a broadcast
/// address.
pub fn set_broadcast(&self, on: bool) -> io::Result<()> {
self.io.get_ref().set_broadcast(on)
self.io.set_broadcast(on)
}
/// Gets the value of the `IP_MULTICAST_LOOP` option for this socket.
@@ -685,7 +652,7 @@ impl UdpSocket {
///
/// [`set_multicast_loop_v4`]: method@Self::set_multicast_loop_v4
pub fn multicast_loop_v4(&self) -> io::Result<bool> {
self.io.get_ref().multicast_loop_v4()
self.io.multicast_loop_v4()
}
/// Sets the value of the `IP_MULTICAST_LOOP` option for this socket.
@@ -696,7 +663,7 @@ impl UdpSocket {
///
/// This may not have any affect on IPv6 sockets.
pub fn set_multicast_loop_v4(&self, on: bool) -> io::Result<()> {
self.io.get_ref().set_multicast_loop_v4(on)
self.io.set_multicast_loop_v4(on)
}
/// Gets the value of the `IP_MULTICAST_TTL` option for this socket.
@@ -705,7 +672,7 @@ impl UdpSocket {
///
/// [`set_multicast_ttl_v4`]: method@Self::set_multicast_ttl_v4
pub fn multicast_ttl_v4(&self) -> io::Result<u32> {
self.io.get_ref().multicast_ttl_v4()
self.io.multicast_ttl_v4()
}
/// Sets the value of the `IP_MULTICAST_TTL` option for this socket.
@@ -718,7 +685,7 @@ impl UdpSocket {
///
/// This may not have any affect on IPv6 sockets.
pub fn set_multicast_ttl_v4(&self, ttl: u32) -> io::Result<()> {
self.io.get_ref().set_multicast_ttl_v4(ttl)
self.io.set_multicast_ttl_v4(ttl)
}
/// Gets the value of the `IPV6_MULTICAST_LOOP` option for this socket.
@@ -727,7 +694,7 @@ impl UdpSocket {
///
/// [`set_multicast_loop_v6`]: method@Self::set_multicast_loop_v6
pub fn multicast_loop_v6(&self) -> io::Result<bool> {
self.io.get_ref().multicast_loop_v6()
self.io.multicast_loop_v6()
}
/// Sets the value of the `IPV6_MULTICAST_LOOP` option for this socket.
@@ -738,7 +705,7 @@ impl UdpSocket {
///
/// This may not have any affect on IPv4 sockets.
pub fn set_multicast_loop_v6(&self, on: bool) -> io::Result<()> {
self.io.get_ref().set_multicast_loop_v6(on)
self.io.set_multicast_loop_v6(on)
}
/// Gets the value of the `IP_TTL` option for this socket.
@@ -761,7 +728,7 @@ impl UdpSocket {
/// # }
/// ```
pub fn ttl(&self) -> io::Result<u32> {
self.io.get_ref().ttl()
self.io.ttl()
}
/// Sets the value for the `IP_TTL` option on this socket.
@@ -783,7 +750,7 @@ impl UdpSocket {
/// # }
/// ```
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
self.io.get_ref().set_ttl(ttl)
self.io.set_ttl(ttl)
}
/// Executes an operation of the `IP_ADD_MEMBERSHIP` type.
@@ -794,7 +761,7 @@ impl UdpSocket {
/// multicast group. If it's equal to `INADDR_ANY` then an appropriate
/// interface is chosen by the system.
pub fn join_multicast_v4(&self, multiaddr: Ipv4Addr, interface: Ipv4Addr) -> io::Result<()> {
self.io.get_ref().join_multicast_v4(&multiaddr, &interface)
self.io.join_multicast_v4(&multiaddr, &interface)
}
/// Executes an operation of the `IPV6_ADD_MEMBERSHIP` type.
@@ -803,7 +770,7 @@ impl UdpSocket {
/// The address must be a valid multicast address, and `interface` is the
/// index of the interface to join/leave (or 0 to indicate any interface).
pub fn join_multicast_v6(&self, multiaddr: &Ipv6Addr, interface: u32) -> io::Result<()> {
self.io.get_ref().join_multicast_v6(multiaddr, interface)
self.io.join_multicast_v6(multiaddr, interface)
}
/// Executes an operation of the `IP_DROP_MEMBERSHIP` type.
@@ -812,7 +779,7 @@ impl UdpSocket {
///
/// [`join_multicast_v4`]: method@Self::join_multicast_v4
pub fn leave_multicast_v4(&self, multiaddr: Ipv4Addr, interface: Ipv4Addr) -> io::Result<()> {
self.io.get_ref().leave_multicast_v4(&multiaddr, &interface)
self.io.leave_multicast_v4(&multiaddr, &interface)
}
/// Executes an operation of the `IPV6_DROP_MEMBERSHIP` type.
@@ -821,7 +788,7 @@ impl UdpSocket {
///
/// [`join_multicast_v6`]: method@Self::join_multicast_v6
pub fn leave_multicast_v6(&self, multiaddr: &Ipv6Addr, interface: u32) -> io::Result<()> {
self.io.get_ref().leave_multicast_v6(multiaddr, interface)
self.io.leave_multicast_v6(multiaddr, interface)
}
/// Returns the value of the `SO_ERROR` option.
@@ -844,7 +811,7 @@ impl UdpSocket {
/// # }
/// ```
pub fn take_error(&self) -> io::Result<Option<io::Error>> {
self.io.get_ref().take_error()
self.io.take_error()
}
}
@@ -862,7 +829,7 @@ impl TryFrom<std::net::UdpSocket> for UdpSocket {
impl fmt::Debug for UdpSocket {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.io.get_ref().fmt(f)
self.io.fmt(f)
}
}
@@ -873,7 +840,7 @@ mod sys {
impl AsRawFd for UdpSocket {
fn as_raw_fd(&self) -> RawFd {
self.io.get_ref().as_raw_fd()
self.io.as_raw_fd()
}
}
}
@@ -885,7 +852,7 @@ mod sys {
impl AsRawSocket for UdpSocket {
fn as_raw_socket(&self) -> RawSocket {
self.io.get_ref().as_raw_socket()
self.io.as_raw_socket()
}
}
}
tokio/src/net/unix/datagram/socket.rs
+20 -18
View File
@@ -1,4 +1,4 @@
use crate::io::PollEvented;
use crate::io::{Interest, PollEvented};
use crate::net::unix::SocketAddr;
use std::convert::TryFrom;
@@ -270,7 +270,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn connect<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
self.io.get_ref().connect(path)
self.io.connect(path)
}
/// Sends data on the socket to the socket's peer.
@@ -301,7 +301,8 @@ impl UnixDatagram {
/// ```
pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
self.io
.async_io(mio::Interest::WRITABLE, |sock| sock.send(buf))
.registration()
.async_io(Interest::WRITABLE, || self.io.send(buf))
.await
}
@@ -330,7 +331,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn try_send(&self, buf: &[u8]) -> io::Result<usize> {
self.io.get_ref().send(buf)
self.io.send(buf)
}
/// Try to send a datagram to the peer without waiting.
@@ -369,7 +370,7 @@ impl UnixDatagram {
where
P: AsRef<Path>,
{
self.io.get_ref().send_to(buf, target)
self.io.send_to(buf, target)
}
/// Receives data from the socket.
@@ -400,7 +401,8 @@ impl UnixDatagram {
/// ```
pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
self.io
.async_io(mio::Interest::READABLE, |sock| sock.recv(buf))
.registration()
.async_io(Interest::READABLE, || self.io.recv(buf))
.await
}
@@ -429,7 +431,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn try_recv(&self, buf: &mut [u8]) -> io::Result<usize> {
self.io.get_ref().recv(buf)
self.io.recv(buf)
}
/// Sends data on the socket to the specified address.
@@ -470,9 +472,8 @@ impl UnixDatagram {
P: AsRef<Path>,
{
self.io
.async_io(mio::Interest::WRITABLE, |sock| {
sock.send_to(buf, target.as_ref())
})
.registration()
.async_io(Interest::WRITABLE, || self.io.send_to(buf, target.as_ref()))
.await
}
@@ -512,7 +513,8 @@ impl UnixDatagram {
pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
let (n, addr) = self
.io
.async_io(mio::Interest::READABLE, |sock| sock.recv_from(buf))
.registration()
.async_io(Interest::READABLE, || self.io.recv_from(buf))
.await?;
Ok((n, SocketAddr(addr)))
@@ -551,7 +553,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn try_recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
let (n, addr) = self.io.get_ref().recv_from(buf)?;
let (n, addr) = self.io.recv_from(buf)?;
Ok((n, SocketAddr(addr)))
}
@@ -596,7 +598,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().local_addr().map(SocketAddr)
self.io.local_addr().map(SocketAddr)
}
/// Returns the address of this socket's peer.
@@ -645,7 +647,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().peer_addr().map(SocketAddr)
self.io.peer_addr().map(SocketAddr)
}
/// Returns the value of the `SO_ERROR` option.
@@ -668,7 +670,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn take_error(&self) -> io::Result<Option<io::Error>> {
self.io.get_ref().take_error()
self.io.take_error()
}
/// Shuts down the read, write, or both halves of this connection.
@@ -704,7 +706,7 @@ impl UnixDatagram {
/// # }
/// ```
pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
self.io.get_ref().shutdown(how)
self.io.shutdown(how)
}
}
@@ -722,12 +724,12 @@ impl TryFrom<std::os::unix::net::UnixDatagram> for UnixDatagram {
impl fmt::Debug for UnixDatagram {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.io.get_ref().fmt(f)
self.io.fmt(f)
}
}
impl AsRawFd for UnixDatagram {
fn as_raw_fd(&self) -> RawFd {
self.io.get_ref().as_raw_fd()
self.io.as_raw_fd()
}
}
tokio/src/net/unix/listener.rs
+11 -21
View File
@@ -1,4 +1,4 @@
use crate::io::PollEvented;
use crate::io::{Interest, PollEvented};
use crate::net::unix::{SocketAddr, UnixStream};
use std::convert::TryFrom;
@@ -90,19 +90,20 @@ impl UnixListener {
/// Returns the local socket address of this listener.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().local_addr().map(SocketAddr)
self.io.local_addr().map(SocketAddr)
}
/// Returns the value of the `SO_ERROR` option.
pub fn take_error(&self) -> io::Result<Option<io::Error>> {
self.io.get_ref().take_error()
self.io.take_error()
}
/// Accepts a new incoming connection to this listener.
pub async fn accept(&self) -> io::Result<(UnixStream, SocketAddr)> {
let (mio, addr) = self
.io
.async_io(mio::Interest::READABLE, |sock| sock.accept())
.registration()
.async_io(Interest::READABLE, || self.io.accept())
.await?;
let addr = SocketAddr(addr);
@@ -119,21 +120,10 @@ impl UnixListener {
/// The caller is responsible to ensure that `poll_accept` is called from a
/// single task. Failing to do this could result in tasks hanging.
pub fn poll_accept(&self, cx: &mut Context<'_>) -> Poll<io::Result<(UnixStream, SocketAddr)>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
match self.io.get_ref().accept() {
Ok((sock, addr)) => {
let addr = SocketAddr(addr);
let sock = UnixStream::new(sock)?;
return Poll::Ready(Ok((sock, addr)));
}
Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
Err(err) => return Err(err).into(),
}
}
let (sock, addr) = ready!(self.io.registration().poll_read_io(cx, || self.io.accept()))?;
let addr = SocketAddr(addr);
let sock = UnixStream::new(sock)?;
Poll::Ready(Ok((sock, addr)))
}
}
@@ -161,12 +151,12 @@ impl TryFrom<std::os::unix::net::UnixListener> for UnixListener {
impl fmt::Debug for UnixListener {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.io.get_ref().fmt(f)
self.io.fmt(f)
}
}
impl AsRawFd for UnixListener {
fn as_raw_fd(&self) -> RawFd {
self.io.get_ref().as_raw_fd()
self.io.as_raw_fd()
}
}
tokio/src/net/unix/stream.rs
+11 -41
View File
@@ -7,7 +7,7 @@ use crate::net::unix::SocketAddr;
use std::convert::TryFrom;
use std::fmt;
use std::io::{self, Read, Write};
use std::io;
use std::net::Shutdown;
use std::os::unix::io::{AsRawFd, RawFd};
use std::os::unix::net;
@@ -39,7 +39,7 @@ impl UnixStream {
let stream = mio::net::UnixStream::connect(path)?;
let stream = UnixStream::new(stream)?;
poll_fn(|cx| stream.io.poll_write_ready(cx)).await?;
poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;
Ok(stream)
}
@@ -84,12 +84,12 @@ impl UnixStream {
/// Returns the socket address of the local half of this connection.
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().local_addr().map(SocketAddr)
self.io.local_addr().map(SocketAddr)
}
/// Returns the socket address of the remote half of this connection.
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().peer_addr().map(SocketAddr)
self.io.peer_addr().map(SocketAddr)
}
/// Returns effective credentials of the process which called `connect` or `pair`.
@@ -99,7 +99,7 @@ impl UnixStream {
/// Returns the value of the `SO_ERROR` option.
pub fn take_error(&self) -> io::Result<Option<io::Error>> {
self.io.get_ref().take_error()
self.io.take_error()
}
/// Shuts down the read, write, or both halves of this connection.
@@ -108,7 +108,7 @@ impl UnixStream {
/// specified portions to immediately return with an appropriate value
/// (see the documentation of `Shutdown`).
pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
self.io.get_ref().shutdown(how)
self.io.shutdown(how)
}
// These lifetime markers also appear in the generated documentation, and make
@@ -199,29 +199,8 @@ impl UnixStream {
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
loop {
let ev = ready!(self.io.poll_read_ready(cx))?;
// Safety: `UnixStream::read` will not peek at the maybe uinitialized bytes.
let b = unsafe {
&mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
};
match self.io.get_ref().read(b) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
Ok(n) => {
// Safety: We trust `UnixStream::read` to have filled up `n` bytes
// in the buffer.
unsafe {
buf.assume_init(n);
}
buf.advance(n);
return Poll::Ready(Ok(()));
}
Err(e) => return Poll::Ready(Err(e)),
}
}
// Safety: `UdpStream::read` correctly handles reads into uninitialized memory
unsafe { self.io.poll_read(cx, buf) }
}
pub(crate) fn poll_write_priv(
@@ -229,27 +208,18 @@ impl UnixStream {
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
loop {
let ev = ready!(self.io.poll_write_ready(cx))?;
match self.io.get_ref().write(buf) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
self.io.clear_readiness(ev);
}
x => return Poll::Ready(x),
}
}
self.io.poll_write(cx, buf)
}
}
impl fmt::Debug for UnixStream {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.io.get_ref().fmt(f)
self.io.fmt(f)
}
}
impl AsRawFd for UnixStream {
fn as_raw_fd(&self) -> RawFd {
self.io.get_ref().as_raw_fd()
self.io.as_raw_fd()
}
}
tokio/src/process/mod.rs
+18 -16
View File
@@ -1038,39 +1038,41 @@ pub struct ChildStderr {
impl AsyncWrite for ChildStdin {
fn poll_write(
mut self: Pin<&mut Self>,
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
Pin::new(&mut self.inner).poll_write(cx, buf)
self.inner.poll_write(cx, buf)
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.inner).poll_flush(cx)
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.inner).poll_shutdown(cx)
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
impl AsyncRead for ChildStdout {
fn poll_read(
mut self: Pin<&mut Self>,
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.inner).poll_read(cx, buf)
// Safety: pipes support reading into uninitialized memory
unsafe { self.inner.poll_read(cx, buf) }
}
}
impl AsyncRead for ChildStderr {
fn poll_read(
mut self: Pin<&mut Self>,
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.inner).poll_read(cx, buf)
// Safety: pipes support reading into uninitialized memory
unsafe { self.inner.poll_read(cx, buf) }
}
}
@@ -1082,19 +1084,19 @@ mod sys {
impl AsRawFd for ChildStdin {
fn as_raw_fd(&self) -> RawFd {
self.inner.get_ref().as_raw_fd()
self.inner.as_raw_fd()
}
}
impl AsRawFd for ChildStdout {
fn as_raw_fd(&self) -> RawFd {
self.inner.get_ref().as_raw_fd()
self.inner.as_raw_fd()
}
}
impl AsRawFd for ChildStderr {
fn as_raw_fd(&self) -> RawFd {
self.inner.get_ref().as_raw_fd()
self.inner.as_raw_fd()
}
}
}
@@ -1107,19 +1109,19 @@ mod sys {
impl AsRawHandle for ChildStdin {
fn as_raw_handle(&self) -> RawHandle {
self.inner.get_ref().as_raw_handle()
self.inner.as_raw_handle()
}
}
impl AsRawHandle for ChildStdout {
fn as_raw_handle(&self) -> RawHandle {
self.inner.get_ref().as_raw_handle()
self.inner.as_raw_handle()
}
}
impl AsRawHandle for ChildStderr {
fn as_raw_handle(&self) -> RawHandle {
self.inner.get_ref().as_raw_handle()
self.inner.as_raw_handle()
}
}
}
tokio/src/process/unix/mod.rs
+26 -32
View File
@@ -37,9 +37,10 @@ use crate::signal::unix::{signal, Signal, SignalKind};
use mio::event::Source;
use mio::unix::SourceFd;
use std::fmt;
use std::fs::File;
use std::future::Future;
use std::io;
use std::os::unix::io::{AsRawFd, RawFd};
use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
use std::pin::Pin;
use std::process::{Child as StdChild, ExitStatus};
use std::task::Context;
@@ -141,45 +142,35 @@ impl Future for Child {
}
#[derive(Debug)]
pub(crate) struct Fd<T> {
inner: T,
pub(crate) struct Pipe {
// Actually a pipe and not a File. However, we are reusing `File` to get
// close on drop. This is a similar trick as `mio`.
fd: File,
}
impl<T> io::Read for Fd<T>
where
T: io::Read,
{
impl<'a> io::Read for &'a Pipe {
fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
self.inner.read(bytes)
(&self.fd).read(bytes)
}
}
impl<T> io::Write for Fd<T>
where
T: io::Write,
{
impl<'a> io::Write for &'a Pipe {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.inner.write(bytes)
(&self.fd).write(bytes)
}
fn flush(&mut self) -> io::Result<()> {
self.inner.flush()
(&self.fd).flush()
}
}
impl<T> AsRawFd for Fd<T>
where
T: AsRawFd,
{
impl AsRawFd for Pipe {
fn as_raw_fd(&self) -> RawFd {
self.inner.as_raw_fd()
self.fd.as_raw_fd()
}
}
impl<T> Source for Fd<T>
where
T: AsRawFd,
{
impl Source for Pipe {
fn register(
&mut self,
registry: &mio::Registry,
@@ -203,13 +194,13 @@ where
}
}
pub(crate) type ChildStdin = PollEvented<Fd<std::process::ChildStdin>>;
pub(crate) type ChildStdout = PollEvented<Fd<std::process::ChildStdout>>;
pub(crate) type ChildStderr = PollEvented<Fd<std::process::ChildStderr>>;
pub(crate) type ChildStdin = PollEvented<Pipe>;
pub(crate) type ChildStdout = PollEvented<Pipe>;
pub(crate) type ChildStderr = PollEvented<Pipe>;
fn stdio<T>(option: Option<T>) -> io::Result<Option<PollEvented<Fd<T>>>>
fn stdio<T>(option: Option<T>) -> io::Result<Option<PollEvented<Pipe>>>
where
T: AsRawFd,
T: IntoRawFd,
{
let io = match option {
Some(io) => io,
@@ -217,8 +208,8 @@ where
};
// Set the fd to nonblocking before we pass it to the event loop
unsafe {
let fd = io.as_raw_fd();
let fd = unsafe {
let fd = io.into_raw_fd();
let r = libc::fcntl(fd, libc::F_GETFL);
if r == -1 {
return Err(io::Error::last_os_error());
@@ -227,6 +218,9 @@ where
if r == -1 {
return Err(io::Error::last_os_error());
}
}
Ok(Some(PollEvented::new(Fd { inner: io })?))
File::from_raw_fd(fd)
};
Ok(Some(PollEvented::new(Pipe { fd })?))
}
tokio/src/signal/unix/driver.rs
+5 -5
View File
@@ -3,7 +3,7 @@
//! Signal driver
use crate::io::driver::Driver as IoDriver;
use crate::io::PollEvented;
use crate::io::{Interest, PollEvented};
use crate::park::Park;
use crate::signal::registry::globals;
@@ -76,7 +76,7 @@ impl Driver {
let receiver = UnixStream::from_std(original.try_clone()?);
let receiver = PollEvented::new_with_interest_and_handle(
receiver,
mio::Interest::READABLE | mio::Interest::WRITABLE,
Interest::READABLE | Interest::WRITABLE,
park.handle(),
)?;
@@ -103,7 +103,7 @@ impl Driver {
let waker = unsafe { Waker::from_raw(RawWaker::new(ptr::null(), &NOOP_WAKER_VTABLE)) };
let mut cx = Context::from_waker(&waker);
let ev = match self.receiver.poll_read_ready(&mut cx) {
let ev = match self.receiver.registration().poll_read_ready(&mut cx) {
Poll::Ready(Ok(ev)) => ev,
Poll::Ready(Err(e)) => panic!("reactor gone: {}", e),
Poll::Pending => return, // No wake has arrived, bail
@@ -113,7 +113,7 @@ impl Driver {
// if another signal has come in.
let mut buf = [0; 128];
loop {
match self.receiver.get_ref().read(&mut buf) {
match (&*self.receiver).read(&mut buf) {
Ok(0) => panic!("EOF on self-pipe"),
Ok(_) => continue, // Keep reading
Err(e) if e.kind() == io::ErrorKind::WouldBlock => break,
@@ -121,7 +121,7 @@ impl Driver {
}
}
self.receiver.clear_readiness(ev);
self.receiver.registration().clear_readiness(ev);
// Broadcast any signals which were received
globals().broadcast();