perf: eager poll async ops in Isolate (#3046)

1 files changed, 150 insertions, 59 deletions

diff --git a/core/isolate.rs b/core/isolate.rs
index 2f544a20a..ac5c8402a 100644
--- a/core/isolate.rs
+++ b/core/isolate.rs
@@ -315,6 +315,21 @@ impl Isolate {
       PinnedBuf::new(zero_copy_buf),
     );
 
+    let op = match op {
+      Op::Async(mut fut) => {
+        // Tries to greedily poll async ops once. Often they are immediately ready, in
+        // which case they can be turned into a sync op before we return to V8. This
+        // can save a boundary crossing.
+        #[allow(clippy::match_wild_err_arm)]
+        match fut.poll() {
+          Err(_) => panic!("unexpected op error"),
+          Ok(Ready(buf)) => Op::Sync(buf),
+          Ok(NotReady) => Op::Async(fut),
+        }
+      }
+      Op::Sync(buf) => Op::Sync(buf),
+    };
+
     debug_assert_eq!(isolate.shared.size(), 0);
     match op {
       Op::Sync(buf) => {
@@ -748,8 +763,34 @@ pub mod tests {
     )
   }
 
+  struct DelayedFuture {
+    counter: u32,
+    buf: Box<[u8]>,
+  }
+
+  impl DelayedFuture {
+    pub fn new(buf: Box<[u8]>) -> Self {
+      DelayedFuture { counter: 0, buf }
+    }
+  }
+
+  impl Future for DelayedFuture {
+    type Item = Box<[u8]>;
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+      if self.counter > 0 {
+        return Ok(Async::Ready(self.buf.clone()));
+      }
+
+      self.counter += 1;
+      Ok(Async::NotReady)
+    }
+  }
+
   pub enum Mode {
     AsyncImmediate,
+    AsyncDelayed,
     OverflowReqSync,
     OverflowResSync,
     OverflowReqAsync,
@@ -772,6 +813,12 @@ pub mod tests {
             let buf = vec![43u8, 0, 0, 0].into_boxed_slice();
             Op::Async(Box::new(futures::future::ok(buf)))
           }
+          Mode::AsyncDelayed => {
+            assert_eq!(control.len(), 1);
+            assert_eq!(control[0], 42);
+            let buf = vec![43u8, 0, 0, 0].into_boxed_slice();
+            Op::Async(Box::new(DelayedFuture::new(buf)))
+          }
           Mode::OverflowReqSync => {
             assert_eq!(control.len(), 100 * 1024 * 1024);
             let buf = vec![43u8, 0, 0, 0].into_boxed_slice();
@@ -789,7 +836,7 @@ pub mod tests {
           Mode::OverflowReqAsync => {
             assert_eq!(control.len(), 100 * 1024 * 1024);
             let buf = vec![43u8, 0, 0, 0].into_boxed_slice();
-            Op::Async(Box::new(futures::future::ok(buf)))
+            Op::Async(Box::new(DelayedFuture::new(buf)))
           }
           Mode::OverflowResAsync => {
             assert_eq!(control.len(), 1);
@@ -798,7 +845,7 @@ pub mod tests {
             vec.resize(100 * 1024 * 1024, 0);
             vec[0] = 4;
             let buf = vec.into_boxed_slice();
-            Op::Async(Box::new(futures::future::ok(buf)))
+            Op::Async(Box::new(DelayedFuture::new(buf)))
           }
         }
       };
@@ -892,21 +939,65 @@ pub mod tests {
       js_check(isolate.execute(
         "setup2.js",
         r#"
-        let nrecv = 0;
-        Deno.core.setAsyncHandler((opId, buf) => {
-          nrecv++;
-        });
-        "#,
+         let nrecv = 0;
+         Deno.core.setAsyncHandler((opId, buf) => {
+           nrecv++;
+         });
+         "#,
       ));
       assert_eq!(dispatch_count.load(Ordering::Relaxed), 0);
       js_check(isolate.execute(
         "check1.js",
         r#"
-        assert(nrecv == 0);
-        let control = new Uint8Array([42]);
-        Deno.core.send(1, control);
-        assert(nrecv == 0);
-        "#,
+         assert(nrecv == 0);
+         let control = new Uint8Array([42]);
+         const res1 = Deno.core.send(1, control);
+         assert(res1);
+         assert(nrecv == 0);
+         "#,
+      ));
+      assert_eq!(dispatch_count.load(Ordering::Relaxed), 1);
+      assert_eq!(dispatch_count.load(Ordering::Relaxed), 1);
+      js_check(isolate.execute(
+        "check2.js",
+        r#"
+         assert(nrecv == 0);
+         Deno.core.send(1, control);
+         assert(nrecv == 0);
+         "#,
+      ));
+      assert_eq!(dispatch_count.load(Ordering::Relaxed), 2);
+      assert_eq!(dispatch_count.load(Ordering::Relaxed), 2);
+      assert_eq!(Async::Ready(()), isolate.poll().unwrap());
+      js_check(isolate.execute("check3.js", "assert(nrecv == 0)"));
+      // We are idle, so the next poll should be the last.
+      assert_eq!(Async::Ready(()), isolate.poll().unwrap());
+    });
+  }
+
+  #[test]
+  fn test_poll_async_delayed_ops() {
+    run_in_task(|| {
+      let (mut isolate, dispatch_count) = setup(Mode::AsyncDelayed);
+
+      js_check(isolate.execute(
+        "setup2.js",
+        r#"
+         let nrecv = 0;
+         Deno.core.setAsyncHandler((opId, buf) => {
+           nrecv++;
+         });
+         "#,
+      ));
+      assert_eq!(dispatch_count.load(Ordering::Relaxed), 0);
+      js_check(isolate.execute(
+        "check1.js",
+        r#"
+         assert(nrecv == 0);
+         let control = new Uint8Array([42]);
+         Deno.core.send(1, control);
+         assert(nrecv == 0);
+         "#,
       ));
       assert_eq!(dispatch_count.load(Ordering::Relaxed), 1);
       assert_eq!(Async::Ready(()), isolate.poll().unwrap());
@@ -914,10 +1005,10 @@ pub mod tests {
       js_check(isolate.execute(
         "check2.js",
         r#"
-        assert(nrecv == 1);
-        Deno.core.send(1, control);
-        assert(nrecv == 1);
-        "#,
+         assert(nrecv == 1);
+         Deno.core.send(1, control);
+         assert(nrecv == 1);
+         "#,
       ));
       assert_eq!(dispatch_count.load(Ordering::Relaxed), 2);
       assert_eq!(Async::Ready(()), isolate.poll().unwrap());
@@ -1235,20 +1326,20 @@ pub mod tests {
       js_check(isolate.execute(
         "overflow_req_async.js",
         r#"
-        let asyncRecv = 0;
-        Deno.core.setAsyncHandler((opId, buf) => {
-          assert(opId == 1);
-          assert(buf.byteLength === 4);
-          assert(buf[0] === 43);
-          asyncRecv++;
-        });
-        // Large message that will overflow the shared space.
-        let control = new Uint8Array(100 * 1024 * 1024);
-        let response = Deno.core.dispatch(1, control);
-        // Async messages always have null response.
-        assert(response == null);
-        assert(asyncRecv == 0);
-        "#,
+         let asyncRecv = 0;
+         Deno.core.setAsyncHandler((opId, buf) => {
+           assert(opId == 1);
+           assert(buf.byteLength === 4);
+           assert(buf[0] === 43);
+           asyncRecv++;
+         });
+         // Large message that will overflow the shared space.
+         let control = new Uint8Array(100 * 1024 * 1024);
+         let response = Deno.core.dispatch(1, control);
+         // Async messages always have null response.
+         assert(response == null);
+         assert(asyncRecv == 0);
+         "#,
       ));
       assert_eq!(dispatch_count.load(Ordering::Relaxed), 1);
       assert_eq!(Async::Ready(()), js_check(isolate.poll()));
@@ -1265,19 +1356,19 @@ pub mod tests {
       js_check(isolate.execute(
         "overflow_res_async.js",
         r#"
-        let asyncRecv = 0;
-        Deno.core.setAsyncHandler((opId, buf) => {
-          assert(opId == 1);
-          assert(buf.byteLength === 100 * 1024 * 1024);
-          assert(buf[0] === 4);
-          asyncRecv++;
-        });
-        // Large message that will overflow the shared space.
-        let control = new Uint8Array([42]);
-        let response = Deno.core.dispatch(1, control);
-        assert(response == null);
-        assert(asyncRecv == 0);
-        "#,
+         let asyncRecv = 0;
+         Deno.core.setAsyncHandler((opId, buf) => {
+           assert(opId == 1);
+           assert(buf.byteLength === 100 * 1024 * 1024);
+           assert(buf[0] === 4);
+           asyncRecv++;
+         });
+         // Large message that will overflow the shared space.
+         let control = new Uint8Array([42]);
+         let response = Deno.core.dispatch(1, control);
+         assert(response == null);
+         assert(asyncRecv == 0);
+         "#,
       ));
       assert_eq!(dispatch_count.load(Ordering::Relaxed), 1);
       poll_until_ready(&mut isolate, 3).unwrap();
@@ -1294,22 +1385,22 @@ pub mod tests {
       js_check(isolate.execute(
         "overflow_res_multiple_dispatch_async.js",
         r#"
-        let asyncRecv = 0;
-        Deno.core.setAsyncHandler((opId, buf) => {
-          assert(opId === 1);
-          assert(buf.byteLength === 100 * 1024 * 1024);
-          assert(buf[0] === 4);
-          asyncRecv++;
-        });
-        // Large message that will overflow the shared space.
-        let control = new Uint8Array([42]);
-        let response = Deno.core.dispatch(1, control);
-        assert(response == null);
-        assert(asyncRecv == 0);
-        // Dispatch another message to verify that pending ops
-        // are done even if shared space overflows
-        Deno.core.dispatch(1, control);
-        "#,
+         let asyncRecv = 0;
+         Deno.core.setAsyncHandler((opId, buf) => {
+           assert(opId === 1);
+           assert(buf.byteLength === 100 * 1024 * 1024);
+           assert(buf[0] === 4);
+           asyncRecv++;
+         });
+         // Large message that will overflow the shared space.
+         let control = new Uint8Array([42]);
+         let response = Deno.core.dispatch(1, control);
+         assert(response == null);
+         assert(asyncRecv == 0);
+         // Dispatch another message to verify that pending ops
+         // are done even if shared space overflows
+         Deno.core.dispatch(1, control);
+         "#,
       ));
       assert_eq!(dispatch_count.load(Ordering::Relaxed), 2);
       poll_until_ready(&mut isolate, 3).unwrap();