refactor: per-worker resource table (#3306)

- removes global `RESOURCE_TABLE` - resource tables are now created per `Worker`
  in `State`
- renames `CliResource` to `StreamResource` and moves all logic related
  to it to `cli/ops/io.rs`
- removes `cli/resources.rs`
- adds `state` argument to `op_read` and `op_write` and consequently adds
  `stateful_minimal_op` to `State`
- IMPORTANT NOTE: workers don't have access to process stdio - this is
  caused by fact that dropping worker would close stdout for process
  (because it's constructed from raw handle, which closes underlying file
  descriptor on drop)

1 files changed, 24 insertions, 0 deletions

diff --git a/cli/state.rs b/cli/state.rs
index edfac72c0..a5e9546b0 100644
--- a/cli/state.rs
+++ b/cli/state.rs
@@ -5,6 +5,7 @@ use crate::global_timer::GlobalTimer;
 use crate::import_map::ImportMap;
 use crate::metrics::Metrics;
 use crate::ops::JsonOp;
+use crate::ops::MinimalOp;
 use crate::permissions::DenoPermissions;
 use crate::worker::Worker;
 use crate::worker::WorkerChannels;
@@ -15,6 +16,7 @@ use deno::Loader;
 use deno::ModuleSpecifier;
 use deno::Op;
 use deno::PinnedBuf;
+use deno::ResourceTable;
 use futures::Future;
 use rand::rngs::StdRng;
 use rand::SeedableRng;
@@ -27,6 +29,7 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::sync::Mutex;
+use std::sync::MutexGuard;
 use std::time::Instant;
 use tokio::sync::mpsc;
 
@@ -52,6 +55,7 @@ pub struct State {
   pub start_time: Instant,
   pub seeded_rng: Option<Mutex<StdRng>>,
   pub include_deno_namespace: bool,
+  pub resource_table: Mutex<ResourceTable>,
 }
 
 impl Clone for ThreadSafeState {
@@ -68,6 +72,10 @@ impl Deref for ThreadSafeState {
 }
 
 impl ThreadSafeState {
+  pub fn lock_resource_table(&self) -> MutexGuard<ResourceTable> {
+    self.resource_table.lock().unwrap()
+  }
+
   /// Wrap core `OpDispatcher` to collect metrics.
   pub fn core_op<D>(
     &self,
@@ -104,6 +112,21 @@ impl ThreadSafeState {
   }
 
   /// This is a special function that provides `state` argument to dispatcher.
+  pub fn stateful_minimal_op<D>(
+    &self,
+    dispatcher: D,
+  ) -> impl Fn(i32, Option<PinnedBuf>) -> Box<MinimalOp>
+  where
+    D: Fn(&ThreadSafeState, i32, Option<PinnedBuf>) -> Box<MinimalOp>,
+  {
+    let state = self.clone();
+
+    move |rid: i32, zero_copy: Option<PinnedBuf>| -> Box<MinimalOp> {
+      dispatcher(&state, rid, zero_copy)
+    }
+  }
+
+  /// This is a special function that provides `state` argument to dispatcher.
   ///
   /// NOTE: This only works with JSON dispatcher.
   /// This is a band-aid for transition to `Isolate.register_op` API as most of our
@@ -220,6 +243,7 @@ impl ThreadSafeState {
       start_time: Instant::now(),
       seeded_rng,
       include_deno_namespace,
+      resource_table: Mutex::new(ResourceTable::default()),
     };
 
     Ok(ThreadSafeState(Arc::new(state)))