Fix bug with CurrentThread::turn (#152)

CurrentThread::turn uses a turn count strategy to allow `turn` to not
run infinitely. Currently, there is a bug where spawned tasks will not
get executed in calls to `turn`.

This patch fixes the bug by correctly setting the turn count for newly
spawned tasks.

src/executor/current_thread/mod.rs

@@ -466,6 +466,11 @@ impl<'a, P: Park> Entered<'a, P> {
     pub fn turn(&mut self, duration: Option<Duration>)
         -> Result<Turn, TurnError>
     {
+        if self.executor.is_idle() {
+            // Nothing to do
+            return Ok(Turn(()));
+        }
+
         if !self.tick() {
             let res = match duration {
                 Some(duration) => self.executor.park.park_timeout(duration),

src/executor/current_thread/scheduler.rs

@@ -168,9 +168,12 @@ where U: Unpark,
     }
 
     pub fn schedule(&mut self, item: Box<Future<Item = (), Error = ()>>) {
+        // Get the current scheduler tick
+        let tick_num = self.inner.tick_num.load(SeqCst);
+
         let node = Arc::new(Node {
             item: UnsafeCell::new(Some(Task::new(item))),
-            notified_at: AtomicUsize::new(0),
+            notified_at: AtomicUsize::new(tick_num),
             next_all: UnsafeCell::new(ptr::null_mut()),
             prev_all: UnsafeCell::new(ptr::null_mut()),
             next_readiness: AtomicPtr::new(ptr::null_mut()),

tests/current_thread.rs

@@ -257,6 +257,42 @@ fn tasks_are_scheduled_fairly() {
     })).unwrap();
 }
 
+#[test]
+fn spawn_and_tick() {
+    let cnt = Rc::new(Cell::new(0));
+    let c = cnt.clone();
+
+    let mut current_thread = CurrentThread::new();
+
+    // Spawn a basic task to get the executor to turn
+    current_thread.spawn(lazy(move || {
+        Ok(())
+    }));
+
+    // Turn once...
+    current_thread.turn(None).unwrap();
+
+    current_thread.spawn(lazy(move || {
+        c.set(1 + c.get());
+
+        // Spawn!
+        current_thread::spawn(lazy(move || {
+            c.set(1 + c.get());
+            Ok::<(), ()>(())
+        }));
+
+        Ok(())
+    }));
+
+    // This does not run the newly spawned thread
+    current_thread.turn(None).unwrap();
+    assert_eq!(1, cnt.get());
+
+    // This runs the newly spawned thread
+    current_thread.turn(None).unwrap();
+    assert_eq!(2, cnt.get());
+}
+
 fn ok() -> future::FutureResult<(), ()> {
     future::ok(())
 }