diff --git a/src/executor/current_thread/scheduler.rs b/src/executor/current_thread/scheduler.rs index 70d6495f4..53c1fc289 100644 --- a/src/executor/current_thread/scheduler.rs +++ b/src/executor/current_thread/scheduler.rs @@ -202,7 +202,8 @@ where U: Unpark, where F: FnMut(&mut Self, &mut Scheduled), { let mut ret = false; - let tick = self.inner.tick_num.fetch_add(1, SeqCst); + let tick = self.inner.tick_num.fetch_add(1, SeqCst) + .wrapping_add(1); loop { let node = match unsafe { self.inner.dequeue(Some(tick)) } { @@ -439,8 +440,18 @@ impl Inner { } if let Some(tick) = tick { - // Only dequeue if the node matches the tick num - if (*tail).notified_at.load(SeqCst) != tick { + let actual = (*tail).notified_at.load(SeqCst); + + // Only dequeue if the node was not scheduled during the current + // tick. + if actual == tick { + // Only doing the check above **should** be enough in + // practice. However, technically there is a potential for + // deadlocking if there are `usize::MAX` ticks while the thread + // scheduling the task is frozen. + // + // If, for some reason, this is not enough, calling `unpark` + // here will resolve the issue. return Dequeue::Empty; } } diff --git a/src/reactor/mod.rs b/src/reactor/mod.rs index 8a22d1a70..9ae52b273 100644 --- a/src/reactor/mod.rs +++ b/src/reactor/mod.rs @@ -256,7 +256,10 @@ impl Reactor { } /// Returns true if the reactor is currently idle. - pub(crate) fn is_idle(&self) -> bool { + /// + /// Idle is defined as all tasks that have been spawned have completed, + /// either successfully or with an error. + pub fn is_idle(&self) -> bool { self.inner.io_dispatch .read().unwrap() .is_empty() @@ -313,9 +316,11 @@ impl Reactor { if let Some(io) = io_dispatch.get(token) { io.readiness.fetch_or(ready2usize(ready), Relaxed); + if ready.is_writable() { io.writer.notify(); } + if !(ready & (!mio::Ready::writable())).is_empty() { io.reader.notify(); } diff --git a/tests/current_thread.rs b/tests/current_thread.rs index 4f63cee5e..2f3b98e5d 100644 --- a/tests/current_thread.rs +++ b/tests/current_thread.rs @@ -258,7 +258,7 @@ fn tasks_are_scheduled_fairly() { } #[test] -fn spawn_and_tick() { +fn spawn_and_turn() { let cnt = Rc::new(Cell::new(0)); let c = cnt.clone(); @@ -293,6 +293,52 @@ fn spawn_and_tick() { assert_eq!(2, cnt.get()); } +#[test] +fn hammer_turn() { + use futures::sync::mpsc; + + const ITER: usize = 100; + const N: usize = 100; + const THREADS: usize = 4; + + for _ in 0..ITER { + let mut ths = vec![]; + + // Add some jitter + for _ in 0..THREADS { + let th = thread::spawn(|| { + let mut current_thread = CurrentThread::new(); + + let (tx, rx) = mpsc::unbounded(); + + current_thread.spawn({ + rx.for_each(|_| { + Ok(()) + }) + .map_err(|e| panic!("err={:?}", e)) + }); + + thread::spawn(move || { + for _ in 0..N { + tx.unbounded_send(()).unwrap(); + thread::yield_now(); + } + }); + + while !current_thread.is_idle() { + current_thread.turn(None).unwrap(); + } + }); + + ths.push(th); + } + + for th in ths { + th.join().unwrap(); + } + } +} + fn ok() -> future::FutureResult<(), ()> { future::ok(()) }