Skip to content

Commit

Permalink
Make CurrentThread::turn() more fair by always parking with 0 timeout…
Browse files Browse the repository at this point in the history
… first

This ensures that all fd-based futures are put into the queue for the
current tick, if the CurrentThread is parking via the Reactor.

Otherwise, if there are queued up futures already, only those would be
polled in the turn. These futures could then notify others/themselves to
have the queue still non-empty on the next turn. Which then potentially
allows the reactor to never be polled, and thus fd-based futures are
never queued up and polled.

Also return in the Turn return value whether any futures were polled at
all, which allows the caller to know if any work was done at all in this
turn and based on that adjust behaviour.

tokio-rs#310
  • Loading branch information
sdroege committed Apr 24, 2018
1 parent 61d635e commit 5b00d26
Showing 1 changed file with 13 additions and 5 deletions.
18 changes: 13 additions & 5 deletions src/executor/current_thread/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -147,9 +147,10 @@ pub struct TaskExecutor {
_p: ::std::marker::PhantomData<Rc<()>>,
}

/// Returned by the `turn` function
/// Returned by the `turn` function. Contains `true` if any futures were
/// polled at all and `false` otherwise.
#[derive(Debug)]
pub struct Turn(());
pub struct Turn(pub bool);

/// A `CurrentThread` instance bound to a supplied execution conext.
pub struct Entered<'a, P: Park + 'a> {
Expand Down Expand Up @@ -480,7 +481,14 @@ impl<'a, P: Park> Entered<'a, P> {
pub fn turn(&mut self, duration: Option<Duration>)
-> Result<Turn, TurnError>
{
if !self.tick() {
let res = self.executor.park.park_timeout(Duration::from_millis(0));
if res.is_err() {
return Err(TurnError { _p: () });
}

let mut polled = self.tick();

if !polled {
let res = match duration {
Some(duration) => self.executor.park.park_timeout(duration),
None => self.executor.park.park(),
Expand All @@ -490,10 +498,10 @@ impl<'a, P: Park> Entered<'a, P> {
return Err(TurnError { _p: () });
}

self.tick();
polled = self.tick();
}

Ok(Turn(()))
Ok(Turn(polled))
}

fn run_timeout2(&mut self, dur: Option<Duration>)
Expand Down

0 comments on commit 5b00d26

Please sign in to comment.