Prevent dangling pointers being left in the system's polling mechanim and reduce the amount of unsafe code in the library (#76)

* Prevent dangling pointers being left in the system's polling mechanism and
reduce the amount of unsafe code in the library.

Previously, dropping a source without unregistering it risked leaving a
dangling pointer to token memory in the underlying polling mechanism. Now, the
Poller type is responsible for keeping track of what data the system's poller
(eg. epoll) is still pointing to, and maintaining ownership of it.

This commit also slightly reduces the amount of unsafe code related to pointer
dereferences and adds some notes justifying the safety of these operations.

Co-authored-by: Victor Berger <[email protected]>

Author: detly

Cargo.toml

@@ -21,6 +21,7 @@ futures-util = { version = "0.3.5", optional = true, default-features = false, f
 futures-io = { version = "0.3.5", optional = true }
 slotmap = "1.0"
 thiserror = "1.0"
+vec_map = "0.8.2"
 
 [dev-dependencies]
 futures = "0.3.5"

src/io.rs

@@ -54,14 +54,14 @@ impl<'l, F: AsRawFd> Async<'l, F> {
         // register in the loop
         let dispatcher = Rc::new(RefCell::new(IoDispatcher {
             fd: rawfd,
-            token: Box::new(Token::invalid()),
+            token: None,
             waker: None,
             is_registered: false,
             interest: Interest::EMPTY,
             last_readiness: Readiness::EMPTY,
         }));
         let key = inner.sources.borrow_mut().insert(dispatcher.clone());
-        *(dispatcher.borrow_mut().token) = Token { key, sub_id: 0 };
+        dispatcher.borrow_mut().token = Some(Token { key, sub_id: 0 });
         inner.register(&dispatcher)?;
 
         // Straightforward casting would require us to add the bound `Data: 'l` but we don't actually need it
@@ -173,30 +173,30 @@ trait IoLoopInner {
 impl<'l, Data> IoLoopInner for LoopInner<'l, Data> {
     fn register(&self, dispatcher: &RefCell<IoDispatcher>) -> crate::Result<()> {
         let disp = dispatcher.borrow();
-        unsafe {
-            self.poll.borrow_mut().register(
-                disp.fd,
-                Interest::EMPTY,
-                Mode::OneShot,
-                &*disp.token as *const _,
-            )
-        }
+        self.poll.borrow_mut().register(
+            disp.fd,
+            Interest::EMPTY,
+            Mode::OneShot,
+            disp.token.expect("No token for IO dispatcher"),
+        )
     }
 
     fn reregister(&self, dispatcher: &RefCell<IoDispatcher>) -> crate::Result<()> {
         let disp = dispatcher.borrow();
-        unsafe {
-            self.poll.borrow_mut().reregister(
-                disp.fd,
-                disp.interest,
-                Mode::OneShot,
-                &*disp.token as *const _,
-            )
-        }
+        self.poll.borrow_mut().reregister(
+            disp.fd,
+            disp.interest,
+            Mode::OneShot,
+            disp.token.expect("No token for IO dispatcher"),
+        )
     }
 
     fn kill(&self, dispatcher: &RefCell<IoDispatcher>) {
-        let key = dispatcher.borrow().token.key;
+        let key = dispatcher
+            .borrow()
+            .token
+            .expect("No token for IO dispatcher")
+            .key;
         let _source = self
             .sources
             .borrow_mut()
@@ -207,7 +207,7 @@ impl<'l, Data> IoLoopInner for LoopInner<'l, Data> {
 
 struct IoDispatcher {
     fd: RawFd,
-    token: Box<Token>,
+    token: Option<Token>,
     waker: Option<Waker>,
     is_registered: bool,
     interest: Interest,

src/sources/generic.rs

@@ -53,7 +53,10 @@ pub struct Generic<F: AsRawFd, E = std::io::Error> {
     pub interest: Interest,
     /// The programmed mode
     pub mode: Mode,
-    token: Box<Token>,
+
+    // This token is used by the event loop logic to look up this source when an
+    // event occurs.
+    token: Option<Token>,
 
     // This allows us to make the associated error and return types generic.
     _error_type: PhantomData<E>,
@@ -67,7 +70,7 @@ impl<F: AsRawFd> Generic<F, std::io::Error> {
             file,
             interest,
             mode,
-            token: Box::new(Token::invalid()),
+            token: None,
             _error_type: PhantomData::default(),
         }
     }
@@ -78,7 +81,7 @@ impl<F: AsRawFd> Generic<F, std::io::Error> {
             file,
             interest,
             mode,
-            token: Box::new(Token::invalid()),
+            token: None,
             _error_type: PhantomData::default(),
         }
     }
@@ -110,23 +113,20 @@ where
     where
         C: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
     {
-        if token != *self.token {
+        // If the token is invalid or not ours, skip processing.
+        if self.token != Some(token) {
             return Ok(PostAction::Continue);
         }
+
         callback(readiness, &mut self.file)
     }
 
     fn register(&mut self, poll: &mut Poll, token_factory: &mut TokenFactory) -> crate::Result<()> {
-        let token = Box::new(token_factory.token());
-        unsafe {
-            poll.register(
-                self.file.as_raw_fd(),
-                self.interest,
-                self.mode,
-                &*token as *const _,
-            )?;
-        }
-        self.token = token;
+        let token = token_factory.token();
+
+        poll.register(self.file.as_raw_fd(), self.interest, self.mode, token)?;
+
+        self.token = Some(token);
         Ok(())
     }
 
@@ -135,22 +135,17 @@ where
         poll: &mut Poll,
         token_factory: &mut TokenFactory,
     ) -> crate::Result<()> {
-        let token = Box::new(token_factory.token());
-        unsafe {
-            poll.reregister(
-                self.file.as_raw_fd(),
-                self.interest,
-                self.mode,
-                &*token as *const _,
-            )?;
-        }
-        self.token = token;
+        let token = token_factory.token();
+
+        poll.reregister(self.file.as_raw_fd(), self.interest, self.mode, token)?;
+
+        self.token = Some(token);
         Ok(())
     }
 
     fn unregister(&mut self, poll: &mut Poll) -> crate::Result<()> {
         poll.unregister(self.file.as_raw_fd())?;
-        self.token = Box::new(Token::invalid());
+        self.token = None;
         Ok(())
     }
 }

src/sys/epoll.rs

@@ -48,15 +48,23 @@ impl Epoll {
         let events = buffer
             .iter()
             .take(n_ready)
-            .map(|event| PollEvent {
-                readiness: flags_to_readiness(event.events()),
-                token: unsafe { *(event.data() as usize as *const Token) },
+            .map(|event| {
+                // In C, the underlying data type is a union including a void
+                // pointer; in Rust's FFI bindings, it only exposes the u64. The
+                // round-trip conversion is valid however.
+                let token_ptr = event.data() as usize as *const Token;
+                PollEvent {
+                    readiness: flags_to_readiness(event.events()),
+                    // Why this is safe: it points to memory boxed and owned by
+                    // the parent Poller type.
+                    token: unsafe { *token_ptr },
+                }
             })
             .collect();
         Ok(events)
     }
 
-    pub unsafe fn register(
+    pub fn register(
         &mut self,
         fd: RawFd,
         interest: Interest,
@@ -68,7 +76,7 @@ impl Epoll {
             .map_err(Into::into)
     }
 
-    pub unsafe fn reregister(
+    pub fn reregister(
         &mut self,
         fd: RawFd,
         interest: Interest,

src/sys/kqueue.rs

@@ -43,19 +43,27 @@ impl Kqueue {
         let ret = buffer
             .iter()
             .take(nevents)
-            .map(|event| PollEvent {
-                readiness: Readiness {
-                    readable: event.filter() == Ok(EventFilter::EVFILT_READ),
-                    writable: event.filter() == Ok(EventFilter::EVFILT_WRITE),
-                    error: event.flags().contains(EventFlag::EV_ERROR) && event.data() != 0,
-                },
-                token: unsafe { *(event.udata() as usize as *const Token) },
+            .map(|event| {
+                // The kevent data field in Rust's libc FFI bindings is an
+                // intptr_t, which is specified to allow this kind of
+                // conversion.
+                let token_ptr = event.udata() as usize as *const Token;
+                PollEvent {
+                    readiness: Readiness {
+                        readable: event.filter() == Ok(EventFilter::EVFILT_READ),
+                        writable: event.filter() == Ok(EventFilter::EVFILT_WRITE),
+                        error: event.flags().contains(EventFlag::EV_ERROR) && event.data() != 0,
+                    },
+                    // Why this is safe: it points to memory boxed and owned by
+                    // the parent Poller type.
+                    token: unsafe { *token_ptr },
+                }
             })
             .collect();
         Ok(ret)
     }
 
-    pub unsafe fn register(
+    pub fn register(
         &mut self,
         fd: RawFd,
         interest: Interest,
@@ -65,7 +73,7 @@ impl Kqueue {
         self.reregister(fd, interest, mode, token)
     }
 
-    pub unsafe fn reregister(
+    pub fn reregister(
         &mut self,
         fd: RawFd,
         interest: Interest,