refactor: fixes for futures (#3363)

After landing #3358 the benchmarks exploded indicating problems with workers and deno_core_http_bench.

This PR dramatically fixes thread/syscall count that showed up on benchmarks. Thread count is not back to previous levels but difference went from hundreds/thousands to about ~50.

4 files changed, 17 insertions, 22 deletions

diff --git a/cli/compilers/ts.rs b/cli/compilers/ts.rs
index 34bf74ab1..13823b24f 100644
--- a/cli/compilers/ts.rs
+++ b/cli/compilers/ts.rs
@@ -272,8 +272,8 @@ impl TsCompiler {
 
     let worker = TsCompiler::setup_worker(global_state.clone());
     let worker_ = worker.clone();
+    worker.post_message(req_msg).unwrap();
     let first_msg_fut = async move {
-      worker.post_message(req_msg).await.unwrap();
       let result = worker.await;
       if let Err(err) = result {
         // TODO(ry) Need to forward the error instead of exiting.
@@ -382,8 +382,8 @@ impl TsCompiler {
       .add("Compile", &module_url.to_string());
     let global_state_ = global_state.clone();
 
+    worker.post_message(req_msg).unwrap();
     let first_msg_fut = async move {
-      worker.post_message(req_msg).await.unwrap();
       let result = worker.await;
       if let Err(err) = result {
         // TODO(ry) Need to forward the error instead of exiting.
diff --git a/cli/compilers/wasm.rs b/cli/compilers/wasm.rs
index 30a171db4..882e28e43 100644
--- a/cli/compilers/wasm.rs
+++ b/cli/compilers/wasm.rs
@@ -86,14 +86,13 @@ impl WasmCompiler {
     let worker_ = worker.clone();
     let url = source_file.url.clone();
 
+    let _res = worker.post_message(
+      serde_json::to_string(&base64_data)
+        .unwrap()
+        .into_boxed_str()
+        .into_boxed_bytes(),
+    );
     let fut = worker
-      .post_message(
-        serde_json::to_string(&base64_data)
-          .unwrap()
-          .into_boxed_str()
-          .into_boxed_bytes(),
-      )
-      .then(move |_| worker)
       .then(move |result| {
         if let Err(err) = result {
           // TODO(ry) Need to forward the error instead of exiting.
diff --git a/cli/ops/workers.rs b/cli/ops/workers.rs
index ee60c6824..42f93ec57 100644
--- a/cli/ops/workers.rs
+++ b/cli/ops/workers.rs
@@ -271,7 +271,8 @@ fn op_host_post_message(
   let mut table = state.workers.lock().unwrap();
   // TODO: don't return bad resource anymore
   let worker = table.get_mut(&id).ok_or_else(bad_resource)?;
-  tokio_util::block_on(worker.post_message(msg).boxed())
+  worker
+    .post_message(msg)
     .map_err(|e| DenoError::new(ErrorKind::Other, e.to_string()))?;
   Ok(JsonOp::Sync(json!({})))
 }
diff --git a/cli/worker.rs b/cli/worker.rs
index aca822888..d5cc801d8 100644
--- a/cli/worker.rs
+++ b/cli/worker.rs
@@ -144,17 +144,12 @@ impl Worker {
   /// Post message to worker as a host.
   ///
   /// This method blocks current thread.
-  pub fn post_message(
-    self: &Self,
-    buf: Buf,
-  ) -> impl Future<Output = Result<(), ErrBox>> {
+  pub fn post_message(self: &Self, buf: Buf) -> Result<(), ErrBox> {
     let channels = self.external_channels.lock().unwrap();
     let mut sender = channels.sender.clone();
-    async move {
-      let result = sender.send(buf).map_err(ErrBox::from).await;
-      drop(sender);
-      result
-    }
+    futures::executor::block_on(sender.send(buf))
+      .map(|_| ())
+      .map_err(ErrBox::from)
   }
 
   /// Get message from worker as a host.
@@ -396,7 +391,7 @@ mod tests {
 
       let msg = json!("hi").to_string().into_boxed_str().into_boxed_bytes();
 
-      let r = futures::executor::block_on(worker_.post_message(msg).boxed());
+      let r = worker_.post_message(msg);
       assert!(r.is_ok());
 
       let maybe_msg =
@@ -409,7 +404,7 @@ mod tests {
         .to_string()
         .into_boxed_str()
         .into_boxed_bytes();
-      let r = futures::executor::block_on(worker_.post_message(msg).boxed());
+      let r = worker_.post_message(msg);
       assert!(r.is_ok());
     })
   }
@@ -439,7 +434,7 @@ mod tests {
       );
 
       let msg = json!("hi").to_string().into_boxed_str().into_boxed_bytes();
-      let r = futures::executor::block_on(worker_.post_message(msg));
+      let r = worker_.post_message(msg);
       assert!(r.is_ok());
 
       futures::executor::block_on(worker_future).unwrap();