Switch to polling crate

Author: Stjepan Glavina

Cargo.toml

@@ -19,6 +19,7 @@ futures-lite = "0.1.10"
 libc = "0.2.74"
 once_cell = "1.4.0"
 parking = "2.0.0"
+polling = "0.1.0"
 socket2 = { version = "0.3.12", features = ["pair", "unix"] }
 vec-arena = "0.5.0"
 
@@ -28,7 +29,6 @@ winapi = { version = "0.3.9", features = ["ioapiset"] }
 
 [dev-dependencies]
 async-channel = "1.4.0"
-async-dup = "1.2.1"
 blocking = "0.5.0"
 signal-hook = "0.1.16"
 tempfile = "3.1.0"

src/lib.rs

@@ -54,19 +54,19 @@
 
 #![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
 
-use std::fmt::Debug;
 use std::future::Future;
 use std::io::{self, IoSlice, IoSliceMut, Read, Write};
-use std::net::{SocketAddr, TcpListener, TcpStream, UdpSocket};
+use std::mem::ManuallyDrop;
+use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream, UdpSocket};
 #[cfg(windows)]
-use std::os::windows::io::{AsRawSocket, RawSocket};
+use std::os::windows::io::{AsRawSocket, FromRawSocket, RawSocket};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll, Waker};
 use std::time::{Duration, Instant};
 #[cfg(unix)]
 use std::{
-    os::unix::io::{AsRawFd, RawFd},
+    os::unix::io::{AsRawFd, FromRawFd, RawFd},
     os::unix::net::{SocketAddr as UnixSocketAddr, UnixDatagram, UnixListener, UnixStream},
     path::Path,
 };
@@ -79,9 +79,7 @@ use socket2::{Domain, Protocol, Socket, Type};
 use crate::reactor::{Reactor, Source};
 
 pub mod parking;
-
 mod reactor;
-mod sys;
 
 /// A timer that expires after a duration of time.
 ///
@@ -656,7 +654,7 @@ impl<T: Write> AsyncWrite for Async<T> {
     }
 
     fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
-        Poll::Ready(sys::shutdown_write(self.source.raw))
+        Poll::Ready(shutdown_write(self.source.raw))
     }
 }
 
@@ -685,7 +683,7 @@ where
     }
 
     fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
-        Poll::Ready(sys::shutdown_write(self.source.raw))
+        Poll::Ready(shutdown_write(self.source.raw))
     }
 }
 
@@ -1312,3 +1310,25 @@ async fn optimistic(fut: impl Future<Output = io::Result<()>>) -> io::Result<()>
     })
     .await
 }
+
+/// Shuts down the write side of a socket.
+///
+/// If this source is not a socket, the `shutdown()` syscall error is ignored.
+pub fn shutdown_write(#[cfg(unix)] raw: RawFd, #[cfg(windows)] raw: RawSocket) -> io::Result<()> {
+    // This may not be a TCP stream, but that's okay. All we do is attempt a `shutdown()` on the
+    // raw descriptor and ignore errors.
+    let stream = unsafe {
+        ManuallyDrop::new(
+            #[cfg(unix)]
+            TcpStream::from_raw_fd(raw),
+            #[cfg(windows)]
+            TcpStream::from_raw_socket(raw),
+        )
+    };
+
+    // If the socket is a TCP stream, the only actual error can be ENOTCONN.
+    match stream.shutdown(Shutdown::Write) {
+        Err(err) if err.kind() == io::ErrorKind::NotConnected => Err(err),
+        _ => Ok(()),
+    }
+}

src/parking.rs

@@ -73,7 +73,10 @@ pub fn pair() -> (Parker, Unparker) {
 /// Waits for a notification.
 #[derive(Debug)]
 pub struct Parker {
+    /// The inner parker implementation.
     inner: parking::Parker,
+
+    /// Set to `true` when the parker is polling I/O.
     io: Arc<AtomicBool>,
 }
 
@@ -289,7 +292,10 @@ impl Default for Parker {
 /// Notifies a parker.
 #[derive(Clone, Debug)]
 pub struct Unparker {
+    /// The inner unparker implementation.
     inner: parking::Unparker,
+
+    /// Set to `true` when the parker is polling I/O.
     io: Arc<AtomicBool>,
 }
 

src/reactor.rs

@@ -15,10 +15,9 @@ use std::time::{Duration, Instant};
 use concurrent_queue::ConcurrentQueue;
 use futures_lite::*;
 use once_cell::sync::Lazy;
+use polling::{Event, Poller};
 use vec_arena::Arena;
 
-use crate::sys;
-
 /// The reactor.
 ///
 /// There is only one global instance of this type, accessible by [`Reactor::get()`].
@@ -29,8 +28,8 @@ pub(crate) struct Reactor {
     /// Unparks the async-io thread.
     thread_unparker: parking::Unparker,
 
-    /// Raw bindings to epoll/kqueue/wepoll.
-    sys: sys::Reactor,
+    /// Bindings to epoll/kqueue/wepoll.
+    poller: Poller,
 
     /// Ticker bumped before polling.
     ticker: AtomicUsize,
@@ -39,7 +38,7 @@ pub(crate) struct Reactor {
     sources: Mutex<Arena<Arc<Source>>>,
 
     /// Temporary storage for I/O events when polling the reactor.
-    events: Mutex<sys::Events>,
+    events: Mutex<Vec<Event>>,
 
     /// An ordered map of registered timers.
     ///
@@ -74,10 +73,10 @@ impl Reactor {
             Reactor {
                 parker_count: AtomicUsize::new(0),
                 thread_unparker: unparker,
-                sys: sys::Reactor::new().expect("cannot initialize I/O event notification"),
+                poller: Poller::new().expect("cannot initialize I/O event notification"),
                 ticker: AtomicUsize::new(0),
                 sources: Mutex::new(Arena::new()),
-                events: Mutex::new(sys::Events::new()),
+                events: Mutex::new(Vec::new()),
                 timers: Mutex::new(BTreeMap::new()),
                 timer_ops: ConcurrentQueue::bounded(1000),
             }
@@ -143,7 +142,7 @@ impl Reactor {
 
     /// Notifies the thread blocked on the reactor.
     pub(crate) fn notify(&self) {
-        self.sys.notify().expect("failed to notify reactor");
+        self.poller.notify().expect("failed to notify reactor");
     }
 
     /// Registers an I/O source in the reactor.
@@ -153,7 +152,7 @@ impl Reactor {
         #[cfg(windows)] raw: RawSocket,
     ) -> io::Result<Arc<Source>> {
         // Register the file descriptor.
-        self.sys.insert(raw)?;
+        self.poller.insert(raw)?;
 
         // Create an I/O source for this file descriptor.
         let mut sources = self.sources.lock().unwrap();
@@ -177,7 +176,7 @@ impl Reactor {
     pub(crate) fn remove_io(&self, source: &Source) -> io::Result<()> {
         let mut sources = self.sources.lock().unwrap();
         sources.remove(source.key);
-        self.sys.remove(source.raw)
+        self.poller.remove(source.raw)
     }
 
     /// Registers a timer in the reactor.
@@ -287,7 +286,7 @@ impl Reactor {
 /// A lock on the reactor.
 pub(crate) struct ReactorLock<'a> {
     reactor: &'a Reactor,
-    events: MutexGuard<'a, sys::Events>,
+    events: MutexGuard<'a, Vec<Event>>,
 }
 
 impl ReactorLock<'_> {
@@ -313,7 +312,7 @@ impl ReactorLock<'_> {
             .wrapping_add(1);
 
         // Block on I/O events.
-        let res = match self.reactor.sys.wait(&mut self.events, timeout) {
+        let res = match self.reactor.poller.wait(&mut self.events, timeout) {
             // No I/O events occurred.
             Ok(0) => {
                 if timeout != Some(Duration::from_secs(0)) {
@@ -350,11 +349,13 @@ impl ReactorLock<'_> {
                         // previously interested in both readability and
                         // writability, but only one of them was emitted.
                         if !(w.writers.is_empty() && w.readers.is_empty()) {
-                            self.reactor.sys.interest(
+                            self.reactor.poller.interest(
                                 source.raw,
-                                source.key,
-                                !w.readers.is_empty(),
-                                !w.writers.is_empty(),
+                                Event {
+                                    key: source.key,
+                                    readable: !w.readers.is_empty(),
+                                    writable: !w.writers.is_empty(),
+                                },
                             )?;
                         }
                     }
@@ -442,9 +443,14 @@ impl Source {
 
             // If there are no other readers, re-register in the reactor.
             if w.readers.is_empty() {
-                Reactor::get()
-                    .sys
-                    .interest(self.raw, self.key, true, !w.writers.is_empty())?;
+                Reactor::get().poller.interest(
+                    self.raw,
+                    Event {
+                        key: self.key,
+                        readable: true,
+                        writable: !w.writers.is_empty(),
+                    },
+                )?;
             }
 
             // Register the current task's waker if not present already.
@@ -483,9 +489,14 @@ impl Source {
 
             // If there are no other writers, re-register in the reactor.
             if w.writers.is_empty() {
-                Reactor::get()
-                    .sys
-                    .interest(self.raw, self.key, !w.readers.is_empty(), true)?;
+                Reactor::get().poller.interest(
+                    self.raw,
+                    Event {
+                        key: self.key,
+                        readable: !w.readers.is_empty(),
+                        writable: true,
+                    },
+                )?;
             }
 
             // Register the current task's waker if not present already.