Do not hang in poll if event loop is destroyed

Author: stepancheg

src/event_loop/mod.rs

@@ -37,7 +37,7 @@ pub struct Loop {
     id: usize,
     io: mio::Poll,
     events: mio::Events,
-    tx: Arc<Sender<Message>>,
+    inner: Arc<LoopHandleInner>,
     rx: Receiver<Message>,
     io_dispatch: RefCell<Slab<ScheduledIo, usize>>,
     task_dispatch: RefCell<Slab<ScheduledTask, usize>>,
@@ -62,6 +62,31 @@ pub struct Loop {
     timeouts: RefCell<Slab<(Timeout, TimeoutState), usize>>,
 }
 
+impl Drop for Loop {
+    fn drop(&mut self) {
+        // mark event loop as dead, so all schedule operations will be rejected
+        self.inner.alive.store(0, Ordering::SeqCst);
+
+        // Unpark all tasks.
+        // It has no effect for tasks in this event loop,
+        // however tasks in another executors get an error
+        // when they do `poll` right after wakeup.
+        for io in self.io_dispatch.borrow_mut().iter_mut() {
+            if let Some(ref mut reader) = io.reader {
+                reader.unpark();
+            }
+            if let Some(ref mut writer) = io.writer {
+                writer.unpark();
+            }
+        }
+    }
+}
+
+struct LoopHandleInner {
+    tx: Arc<Sender<Message>>,
+    alive: AtomicUsize, // 1 iff even loop is alive
+}
+
 /// Handle to an event loop, used to construct I/O objects, send messages, and
 /// otherwise interact indirectly with the event loop itself.
 ///
@@ -70,7 +95,7 @@ pub struct Loop {
 #[derive(Clone)]
 pub struct LoopHandle {
     id: usize,
-    tx: Arc<Sender<Message>>,
+    inner: Arc<LoopHandleInner>,
 }
 
 /// A non-sendable handle to an event loop, useful for manufacturing instances
@@ -145,7 +170,10 @@ impl Loop {
             id: NEXT_LOOP_ID.fetch_add(1, Ordering::Relaxed),
             io: io,
             events: mio::Events::with_capacity(1024),
-            tx: Arc::new(tx),
+            inner: Arc::new(LoopHandleInner {
+                tx: Arc::new(tx),
+                alive: AtomicUsize::new(1),
+            }),
             rx: rx,
             io_dispatch: RefCell::new(Slab::with_capacity(SLAB_CAPACITY)),
             task_dispatch: RefCell::new(Slab::with_capacity(SLAB_CAPACITY)),
@@ -169,7 +197,7 @@ impl Loop {
     pub fn handle(&self) -> LoopHandle {
         LoopHandle {
             id: self.id,
-            tx: self.tx.clone(),
+            inner: self.inner.clone(),
         }
     }
 
@@ -516,7 +544,7 @@ impl LoopHandle {
                     lp.notify(msg);
                 }
                 None => {
-                    match self.tx.send(msg) {
+                    match self.inner.tx.send(msg) {
                         Ok(()) => {}
 
                         // This should only happen when there was an error

src/event_loop/source.rs

@@ -75,6 +75,7 @@ impl LoopHandle {
     /// This function will also panic if there is not a currently running future
     /// task.
     pub fn schedule_read(&self, tok: &IoToken) {
+        assert!(self.inner.alive.load(Ordering::SeqCst) == 1, "Loop is destroyed");
         self.send(Message::Schedule(tok.token, task::park(), Direction::Read));
     }