Refactor Worker and ThreadSafeState (#3242)

* Split ThreadSafeState into State and GlobalState. State is a "local"
  state belonging to "Worker" while "GlobalState" is state shared by
  whole program.
* Update "Worker" and ops to use "GlobalState" where applicable
* Move and refactor "WorkerChannels" resource

1 files changed, 51 insertions, 53 deletions

diff --git a/cli/compilers/ts.rs b/cli/compilers/ts.rs
index c2a4ccc4a..3c36f14c4 100644
--- a/cli/compilers/ts.rs
+++ b/cli/compilers/ts.rs
@@ -5,8 +5,8 @@ use crate::diagnostics::Diagnostic;
 use crate::disk_cache::DiskCache;
 use crate::file_fetcher::SourceFile;
 use crate::file_fetcher::SourceFileFetcher;
+use crate::global_state::ThreadSafeGlobalState;
 use crate::msg;
-use crate::resources;
 use crate::source_maps::SourceMapGetter;
 use crate::startup_data;
 use crate::state::*;
@@ -16,7 +16,7 @@ use deno::Buf;
 use deno::ErrBox;
 use deno::ModuleSpecifier;
 use futures::Future;
-use futures::Stream;
+use futures::IntoFuture;
 use regex::Regex;
 use std::collections::HashSet;
 use std::fs;
@@ -222,16 +222,20 @@ impl TsCompiler {
   }
 
   /// Create a new V8 worker with snapshot of TS compiler and setup compiler's runtime.
-  fn setup_worker(state: ThreadSafeState) -> Worker {
+  fn setup_worker(global_state: ThreadSafeGlobalState) -> Worker {
+    let worker_state = ThreadSafeState::new(global_state.clone(), None, true)
+      .expect("Unable to create worker state");
+
     // Count how many times we start the compiler worker.
-    state.metrics.compiler_starts.fetch_add(1, Ordering::SeqCst);
+    global_state
+      .metrics
+      .compiler_starts
+      .fetch_add(1, Ordering::SeqCst);
 
     let mut worker = Worker::new(
       "TS".to_string(),
       startup_data::compiler_isolate_init(),
-      // TODO(ry) Maybe we should use a separate state for the compiler.
-      // as was done previously.
-      state.clone(),
+      worker_state,
     );
     worker.execute("denoMain()").unwrap();
     worker.execute("workerMain()").unwrap();
@@ -241,7 +245,7 @@ impl TsCompiler {
 
   pub fn bundle_async(
     self: &Self,
-    state: ThreadSafeState,
+    global_state: ThreadSafeGlobalState,
     module_name: String,
     out_file: String,
   ) -> impl Future<Item = (), Error = ErrBox> {
@@ -253,25 +257,21 @@ impl TsCompiler {
     let root_names = vec![module_name.clone()];
     let req_msg = req(root_names, self.config.clone(), Some(out_file));
 
-    let worker = TsCompiler::setup_worker(state.clone());
-    let resource = worker.state.resource.clone();
-    let compiler_rid = resource.rid;
-    let first_msg_fut =
-      resources::post_message_to_worker(compiler_rid, req_msg)
-        .expect("Bad compiler rid")
-        .then(move |_| worker)
-        .then(move |result| {
-          if let Err(err) = result {
-            // TODO(ry) Need to forward the error instead of exiting.
-            eprintln!("{}", err.to_string());
-            std::process::exit(1);
-          }
-          debug!("Sent message to worker");
-          let stream_future =
-            resources::get_message_stream_from_worker(compiler_rid)
-              .into_future();
-          stream_future.map(|(f, _rest)| f).map_err(|(f, _rest)| f)
-        });
+    let worker = TsCompiler::setup_worker(global_state.clone());
+    let worker_ = worker.clone();
+    let first_msg_fut = worker
+      .post_message(req_msg)
+      .into_future()
+      .then(move |_| worker)
+      .then(move |result| {
+        if let Err(err) = result {
+          // TODO(ry) Need to forward the error instead of exiting.
+          eprintln!("{}", err.to_string());
+          std::process::exit(1);
+        }
+        debug!("Sent message to worker");
+        worker_.get_message()
+      });
 
     first_msg_fut.map_err(|_| panic!("not handled")).and_then(
       move |maybe_msg: Option<Buf>| {
@@ -312,7 +312,7 @@ impl TsCompiler {
   /// If compilation is required then new V8 worker is spawned with fresh TS compiler.
   pub fn compile_async(
     self: &Self,
-    state: ThreadSafeState,
+    global_state: ThreadSafeGlobalState,
     source_file: &SourceFile,
   ) -> Box<CompiledModuleFuture> {
     if self.has_compiled(&source_file.url) {
@@ -359,28 +359,26 @@ impl TsCompiler {
     let root_names = vec![module_url.to_string()];
     let req_msg = req(root_names, self.config.clone(), None);
 
-    let worker = TsCompiler::setup_worker(state.clone());
-    let compiling_job = state.progress.add("Compile", &module_url.to_string());
-    let state_ = state.clone();
-
-    let resource = worker.state.resource.clone();
-    let compiler_rid = resource.rid;
-    let first_msg_fut =
-      resources::post_message_to_worker(compiler_rid, req_msg)
-        .expect("Bad compiler rid")
-        .then(move |_| worker)
-        .then(move |result| {
-          if let Err(err) = result {
-            // TODO(ry) Need to forward the error instead of exiting.
-            eprintln!("{}", err.to_string());
-            std::process::exit(1);
-          }
-          debug!("Sent message to worker");
-          let stream_future =
-            resources::get_message_stream_from_worker(compiler_rid)
-              .into_future();
-          stream_future.map(|(f, _rest)| f).map_err(|(f, _rest)| f)
-        });
+    let worker = TsCompiler::setup_worker(global_state.clone());
+    let worker_ = worker.clone();
+    let compiling_job = global_state
+      .progress
+      .add("Compile", &module_url.to_string());
+    let global_state_ = global_state.clone();
+
+    let first_msg_fut = worker
+      .post_message(req_msg)
+      .into_future()
+      .then(move |_| worker)
+      .then(move |result| {
+        if let Err(err) = result {
+          // TODO(ry) Need to forward the error instead of exiting.
+          eprintln!("{}", err.to_string());
+          std::process::exit(1);
+        }
+        debug!("Sent message to worker");
+        worker_.get_message()
+      });
 
     let fut = first_msg_fut
       .map_err(|_| panic!("not handled"))
@@ -400,7 +398,7 @@ impl TsCompiler {
       .and_then(move |_| {
         // if we are this far it means compilation was successful and we can
         // load compiled filed from disk
-        state_
+        global_state_
           .ts_compiler
           .get_compiled_module(&source_file_.url)
           .map_err(|e| {
@@ -663,7 +661,7 @@ mod tests {
       source_code: include_bytes!("../tests/002_hello.ts").to_vec(),
     };
 
-    let mock_state = ThreadSafeState::mock(vec![
+    let mock_state = ThreadSafeGlobalState::mock(vec![
       String::from("deno"),
       String::from("hello.js"),
     ]);
@@ -696,7 +694,7 @@ mod tests {
       .unwrap()
       .to_string();
 
-    let state = ThreadSafeState::mock(vec![
+    let state = ThreadSafeGlobalState::mock(vec![
       String::from("deno"),
       p.to_string_lossy().into(),
       String::from("$deno$/bundle.js"),