diff --git a/src/executor/current_thread/mod.rs b/src/executor/current_thread/mod.rs index d4d2ed295..3681e9969 100644 --- a/src/executor/current_thread/mod.rs +++ b/src/executor/current_thread/mod.rs @@ -147,9 +147,18 @@ pub struct TaskExecutor { _p: ::std::marker::PhantomData>, } -/// Returned by the `turn` function +/// Returned by the `turn` function. #[derive(Debug)] -pub struct Turn(()); +pub struct Turn { + polled: bool +} + +impl Turn { + /// `true` if any futures were polled at all and `false` otherwise. + pub fn has_polled(&self) -> bool { + self.polled + } +} /// A `CurrentThread` instance bound to a supplied execution conext. pub struct Entered<'a, P: Park + 'a> { @@ -480,20 +489,22 @@ impl<'a, P: Park> Entered<'a, P> { pub fn turn(&mut self, duration: Option) -> Result { - if !self.tick() { - let res = match duration { + let res = if self.executor.scheduler.has_pending_futures() { + self.executor.park.park_timeout(Duration::from_millis(0)) + } else { + match duration { Some(duration) => self.executor.park.park_timeout(duration), None => self.executor.park.park(), - }; - - if res.is_err() { - return Err(TurnError { _p: () }); } + }; - self.tick(); + if res.is_err() { + return Err(TurnError { _p: () }); } - Ok(Turn(())) + let polled = self.tick(); + + Ok(Turn { polled }) } fn run_timeout2(&mut self, dur: Option) diff --git a/src/executor/current_thread/scheduler.rs b/src/executor/current_thread/scheduler.rs index 351730bf8..3d7600f87 100644 --- a/src/executor/current_thread/scheduler.rs +++ b/src/executor/current_thread/scheduler.rs @@ -196,6 +196,15 @@ where U: Unpark, self.inner.enqueue(ptr); } + /// Returns `true` if there are currently any pending futures + pub fn has_pending_futures(&mut self) -> bool { + // See function definition for why the unsafe is needed and + // correctly used here + unsafe { + self.inner.has_pending_futures() + } + } + /// Advance the scheduler state, returning `true` if any futures were /// processed. /// @@ -439,6 +448,22 @@ impl Inner { } } + /// Returns `true` if there are currently any pending futures + /// + /// See `dequeue` for an explanation why this function is unsafe. + unsafe fn has_pending_futures(&self) -> bool { + let tail = *self.tail_readiness.get(); + let next = (*tail).next_readiness.load(Acquire); + + if tail == self.stub() { + if next.is_null() { + return false; + } + } + + true + } + /// The dequeue function from the 1024cores intrusive MPSC queue algorithm /// /// Note that this unsafe as it required mutual exclusion (only one thread diff --git a/tests/current_thread.rs b/tests/current_thread.rs index 1124f727a..1a811cc37 100644 --- a/tests/current_thread.rs +++ b/tests/current_thread.rs @@ -392,6 +392,181 @@ fn hammer_turn() { } } +#[test] +fn turn_has_polled() { + let mut current_thread = CurrentThread::new(); + + // Spawn oneshot receiver + let (sender, receiver) = oneshot::channel::<()>(); + current_thread.spawn(receiver.then(|_| Ok(()))); + + // Turn once... + let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + + // Should've polled the receiver once, but considered it not ready + assert!(res.has_polled()); + + // Turn another time + let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + + // Should've polled nothing, the receiver is not ready yet + assert!(!res.has_polled()); + + // Make the receiver ready + sender.send(()).unwrap(); + + // Turn another time + let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + + // Should've polled the receiver, it's ready now + assert!(res.has_polled()); + + // Now the executor should be empty + assert!(current_thread.is_idle()); + let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + + // So should've polled nothing + assert!(!res.has_polled()); +} + +// Our own mock Park that is never really waiting and the only +// thing it does is to send, on request, something (once) to a onshot +// channel +struct MyPark { + sender: Option>, + send_now: Rc>, +} + +struct MyUnpark; + +impl tokio_executor::park::Park for MyPark { + type Unpark = MyUnpark; + type Error = (); + + fn unpark(&self) -> Self::Unpark { + MyUnpark + } + + fn park(&mut self) -> Result<(), Self::Error> { + // If called twice with send_now, this will intentionally panic + if self.send_now.get() { + self.sender.take().unwrap().send(()).unwrap(); + } + + Ok(()) + } + + fn park_timeout(&mut self, _duration: Duration) -> Result<(), Self::Error> { + self.park() + } +} + +impl tokio_executor::park::Unpark for MyUnpark { + fn unpark(&self) {} +} + +#[test] +fn turn_fair() { + let send_now = Rc::new(Cell::new(false)); + + let (sender, receiver) = oneshot::channel::<()>(); + let (sender_2, receiver_2) = oneshot::channel::<()>(); + let (sender_3, receiver_3) = oneshot::channel::<()>(); + + let my_park = MyPark { + sender: Some(sender_3), + send_now: send_now.clone(), + }; + + let mut current_thread = CurrentThread::new_with_park(my_park); + + let receiver_1_done = Rc::new(Cell::new(false)); + let receiver_1_done_clone = receiver_1_done.clone(); + + // Once an item is received on the oneshot channel, it will immediately + // immediately make the second oneshot channel ready + current_thread.spawn(receiver + .map_err(|_| unreachable!()) + .and_then(move |_| { + sender_2.send(()).unwrap(); + receiver_1_done_clone.set(true); + + Ok(()) + }) + ); + + let receiver_2_done = Rc::new(Cell::new(false)); + let receiver_2_done_clone = receiver_2_done.clone(); + + current_thread.spawn(receiver_2 + .map_err(|_| unreachable!()) + .and_then(move |_| { + receiver_2_done_clone.set(true); + Ok(()) + }) + ); + + // The third receiver is only woken up from our Park implementation, it simulates + // e.g. a socket that first has to be polled to know if it is ready now + let receiver_3_done = Rc::new(Cell::new(false)); + let receiver_3_done_clone = receiver_3_done.clone(); + + current_thread.spawn(receiver_3 + .map_err(|_| unreachable!()) + .and_then(move |_| { + receiver_3_done_clone.set(true); + Ok(()) + }) + ); + + // First turn should've polled both and considered them not ready + let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + assert!(res.has_polled()); + + // Next turn should've polled nothing + let res = current_thread.turn(Some(Duration::from_millis(0))).unwrap(); + assert!(!res.has_polled()); + + assert!(!receiver_1_done.get()); + assert!(!receiver_2_done.get()); + assert!(!receiver_3_done.get()); + + // After this the receiver future will wake up the second receiver future, + // so there are pending futures again + sender.send(()).unwrap(); + + // Now the first receiver should be done, the second receiver should be ready + // to be polled again and the socket not yet + let res = current_thread.turn(None).unwrap(); + assert!(res.has_polled()); + + assert!(receiver_1_done.get()); + assert!(!receiver_2_done.get()); + assert!(!receiver_3_done.get()); + + // Now let our park implementation know that it should send something to sender 3 + send_now.set(true); + + // This should resolve the second receiver directly, but also poll the socket + // and read the packet from it. If it didn't do both here, we would handle + // futures that are woken up from the reactor and directly unfairly and would + // favour the ones that are woken up directly. + let res = current_thread.turn(None).unwrap(); + assert!(res.has_polled()); + + assert!(receiver_1_done.get()); + assert!(receiver_2_done.get()); + assert!(receiver_3_done.get()); + + // Don't send again + send_now.set(false); + + // Now we should be idle and turning should not poll anything + assert!(current_thread.is_idle()); + let res = current_thread.turn(None).unwrap(); + assert!(!res.has_polled()); +} + fn ok() -> future::FutureResult<(), ()> { future::ok(()) }