refactor: Use Tokio's single-threaded runtime (#3844)

This change simplifies how we execute V8. Previously V8 Isolates jumped
around threads every time they were woken up. This was overly complex and
potentially hurting performance in a myriad ways. Now isolates run on
their own dedicated thread and never move.

- blocking_json spawns a thread and does not use a thread pool
- op_host_poll_worker and op_host_resume_worker are non-operational
- removes Worker::get_message and Worker::post_message
- ThreadSafeState::workers table contains WorkerChannel entries instead
  of actual Worker instances.
- MainWorker and CompilerWorker are no longer Futures.
- The multi-threaded version of deno_core_http_bench was removed.
- AyncOps no longer need to be Send + Sync

This PR is very large and several tests were disabled to speed
integration:
- installer_test_local_module_run
- installer_test_remote_module_run
- _015_duplicate_parallel_import
- _026_workers

1 files changed, 56 insertions, 43 deletions

diff --git a/cli/compilers/wasm.rs b/cli/compilers/wasm.rs
index f7094724e..f165654ff 100644
--- a/cli/compilers/wasm.rs
+++ b/cli/compilers/wasm.rs
@@ -6,6 +6,7 @@ use crate::file_fetcher::SourceFile;
 use crate::global_state::ThreadSafeGlobalState;
 use crate::startup_data;
 use crate::state::*;
+use deno_core::ErrBox;
 use futures::FutureExt;
 use serde_derive::Deserialize;
 use serde_json;
@@ -70,56 +71,68 @@ impl WasmCompiler {
     source_file: &SourceFile,
   ) -> Pin<Box<CompiledModuleFuture>> {
     let cache = self.cache.clone();
+    let source_file = source_file.clone();
     let maybe_cached = { cache.lock().unwrap().get(&source_file.url).cloned() };
     if let Some(m) = maybe_cached {
       return futures::future::ok(m).boxed();
     }
     let cache_ = self.cache.clone();
 
-    debug!(">>>>> wasm_compile_async START");
-    let base64_data = base64::encode(&source_file.source_code);
-    let worker = WasmCompiler::setup_worker(global_state);
-    let worker_ = worker.clone();
-    let url = source_file.url.clone();
-
-    Box::pin(async move {
-      let _ = worker
-        .post_message(
-          serde_json::to_string(&base64_data)
-            .unwrap()
-            .into_boxed_str()
-            .into_boxed_bytes(),
-        )
-        .await;
-
-      if let Err(err) = worker.await {
-        // TODO(ry) Need to forward the error instead of exiting.
-        eprintln!("{}", err.to_string());
-        std::process::exit(1);
-      }
-      debug!("Sent message to worker");
-      let json_msg = worker_.get_message().await.expect("not handled");
-
-      debug!("Received message from worker");
-      let module_info: WasmModuleInfo =
-        serde_json::from_slice(&json_msg).unwrap();
-      debug!("WASM module info: {:#?}", &module_info);
-      let code = wrap_wasm_code(
-        &base64_data,
-        &module_info.import_list,
-        &module_info.export_list,
-      );
-      debug!("Generated code: {}", &code);
-      let module = CompiledModule {
-        code,
-        name: url.to_string(),
+    let (load_sender, load_receiver) =
+      tokio::sync::oneshot::channel::<Result<CompiledModule, ErrBox>>();
+
+    std::thread::spawn(move || {
+      debug!(">>>>> wasm_compile_async START");
+      let base64_data = base64::encode(&source_file.source_code);
+      let mut worker = WasmCompiler::setup_worker(global_state);
+      let handle = worker.thread_safe_handle();
+      let url = source_file.url.clone();
+
+      let fut = async move {
+        let _ = handle
+          .post_message(
+            serde_json::to_string(&base64_data)
+              .unwrap()
+              .into_boxed_str()
+              .into_boxed_bytes(),
+          )
+          .await;
+
+        if let Err(err) = (&mut *worker).await {
+          load_sender.send(Err(err)).unwrap();
+          return;
+        }
+
+        debug!("Sent message to worker");
+        let json_msg = handle.get_message().await.expect("not handled");
+
+        debug!("Received message from worker");
+        let module_info: WasmModuleInfo =
+          serde_json::from_slice(&json_msg).unwrap();
+
+        debug!("WASM module info: {:#?}", &module_info);
+        let code = wrap_wasm_code(
+          &base64_data,
+          &module_info.import_list,
+          &module_info.export_list,
+        );
+
+        debug!("Generated code: {}", &code);
+        let module = CompiledModule {
+          code,
+          name: url.to_string(),
+        };
+        {
+          cache_.lock().unwrap().insert(url.clone(), module.clone());
+        }
+        debug!("<<<<< wasm_compile_async END");
+        load_sender.send(Ok(module)).unwrap();
       };
-      {
-        cache_.lock().unwrap().insert(url.clone(), module.clone());
-      }
-      debug!("<<<<< wasm_compile_async END");
-      Ok(module)
-    })
+
+      crate::tokio_util::run_basic(fut);
+    });
+    let fut = async { load_receiver.await.unwrap() };
+    fut.boxed_local()
   }
 }