refactor: rewrite ops to use ResourceTable2 (#8512)

This commit migrates all ops to use new resource table
and "AsyncRefCell".

Old implementation of resource table was completely 
removed and all code referencing it was updated to use
new system.

cli/tests/unit/net_test.ts

@@ -21,8 +21,6 @@ unitTest({ perms: { net: true } }, function netTcpListenClose(): void {
 unitTest(
   {
     perms: { net: true },
-    // TODO:
-    ignore: Deno.build.os === "windows",
   },
   function netUdpListenClose(): void {
     const socket = Deno.listenDatagram({
@@ -257,7 +255,7 @@ unitTest(
 );
 
 unitTest(
-  { ignore: Deno.build.os === "windows", perms: { net: true } },
+  { perms: { net: true } },
   async function netUdpSendReceive(): Promise<void> {
     const alice = Deno.listenDatagram({ port: 3500, transport: "udp" });
     assert(alice.addr.transport === "udp");
@@ -287,7 +285,31 @@ unitTest(
 );
 
 unitTest(
-  { ignore: Deno.build.os === "windows", perms: { net: true } },
+  { perms: { net: true } },
+  async function netUdpConcurrentSendReceive(): Promise<void> {
+    const socket = Deno.listenDatagram({ port: 3500, transport: "udp" });
+    assert(socket.addr.transport === "udp");
+    assertEquals(socket.addr.port, 3500);
+    assertEquals(socket.addr.hostname, "127.0.0.1");
+
+    const recvPromise = socket.receive();
+
+    const sendBuf = new Uint8Array([1, 2, 3]);
+    const sendLen = await socket.send(sendBuf, socket.addr);
+    assertEquals(sendLen, 3);
+
+    const [recvBuf, recvAddr] = await recvPromise;
+    assertEquals(recvBuf.length, 3);
+    assertEquals(1, recvBuf[0]);
+    assertEquals(2, recvBuf[1]);
+    assertEquals(3, recvBuf[2]);
+
+    socket.close();
+  },
+);
+
+unitTest(
+  { perms: { net: true } },
   async function netUdpBorrowMutError(): Promise<void> {
     const socket = Deno.listenDatagram({
       port: 4501,
@@ -335,6 +357,34 @@ unitTest(
   },
 );
 
+// TODO(piscisaureus): Enable after Tokio v0.3/v1.0 upgrade.
+unitTest(
+  { ignore: true, perms: { read: true, write: true } },
+  async function netUnixPacketConcurrentSendReceive(): Promise<void> {
+    const filePath = await Deno.makeTempFile();
+    const socket = Deno.listenDatagram({
+      path: filePath,
+      transport: "unixpacket",
+    });
+    assert(socket.addr.transport === "unixpacket");
+    assertEquals(socket.addr.path, filePath);
+
+    const recvPromise = socket.receive();
+
+    const sendBuf = new Uint8Array([1, 2, 3]);
+    const sendLen = await socket.send(sendBuf, socket.addr);
+    assertEquals(sendLen, 3);
+
+    const [recvBuf, recvAddr] = await recvPromise;
+    assertEquals(recvBuf.length, 3);
+    assertEquals(1, recvBuf[0]);
+    assertEquals(2, recvBuf[1]);
+    assertEquals(3, recvBuf[2]);
+
+    socket.close();
+  },
+);
+
 unitTest(
   { perms: { net: true } },
   async function netTcpListenIteratorBreakClosesResource(): Promise<void> {
@@ -385,7 +435,7 @@ unitTest(
 );
 
 unitTest(
-  { ignore: Deno.build.os === "windows", perms: { net: true } },
+  { perms: { net: true } },
   async function netUdpListenCloseWhileIterating(): Promise<void> {
     const socket = Deno.listenDatagram({ port: 8000, transport: "udp" });
     const nextWhileClosing = socket[Symbol.asyncIterator]().next();

core/async_cell.rs

@@ -1,10 +1,14 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 
+use std::any::type_name;
 use std::any::Any;
 use std::borrow::Borrow;
 use std::cell::Cell;
 use std::cell::UnsafeCell;
 use std::collections::VecDeque;
+use std::fmt;
+use std::fmt::Debug;
+use std::fmt::Formatter;
 use std::ops::Deref;
 use std::rc::Rc;
 
@@ -45,6 +49,17 @@ impl<T: 'static> AsyncRefCell<T> {
   pub fn as_ptr(&self) -> *mut T {
     self.value.get()
   }
+
+  pub fn into_inner(self) -> T {
+    assert!(self.borrow_count.get().is_empty());
+    self.value.into_inner()
+  }
+}
+
+impl<T> Debug for AsyncRefCell<T> {
+  fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+    write!(f, "AsyncRefCell<{}>", type_name::<T>())
+  }
 }
 
 impl<T: Default + 'static> Default for AsyncRefCell<T> {

core/examples/http_bench_bin_ops.rs

@@ -170,7 +170,7 @@ fn op_listen(
   let std_listener = std::net::TcpListener::bind(&addr)?;
   std_listener.set_nonblocking(true)?;
   let listener = TcpListener::try_from(std_listener)?;
-  let rid = state.resource_table_2.add(listener);
+  let rid = state.resource_table.add(listener);
   Ok(rid)
 }
 
@@ -181,7 +181,7 @@ fn op_close(
 ) -> Result<u32, Error> {
   debug!("close rid={}", rid);
   state
-    .resource_table_2
+    .resource_table
     .close(rid)
     .map(|_| 0)
     .ok_or_else(bad_resource_id)
@@ -196,11 +196,11 @@ async fn op_accept(
 
   let listener = state
     .borrow()
-    .resource_table_2
+    .resource_table
     .get::<TcpListener>(rid)
     .ok_or_else(bad_resource_id)?;
   let stream = listener.accept().await?;
-  let rid = state.borrow_mut().resource_table_2.add(stream);
+  let rid = state.borrow_mut().resource_table.add(stream);
   Ok(rid)
 }
 
@@ -214,7 +214,7 @@ async fn op_read(
 
   let stream = state
     .borrow()
-    .resource_table_2
+    .resource_table
     .get::<TcpStream>(rid)
     .ok_or_else(bad_resource_id)?;
   stream.read(&mut bufs[0]).await
@@ -230,7 +230,7 @@ async fn op_write(
 
   let stream = state
     .borrow()
-    .resource_table_2
+    .resource_table
     .get::<TcpStream>(rid)
     .ok_or_else(bad_resource_id)?;
   stream.write(&bufs[0]).await

core/examples/http_bench_json_ops.rs

@@ -134,7 +134,7 @@ fn op_listen(
   let std_listener = std::net::TcpListener::bind(&addr)?;
   std_listener.set_nonblocking(true)?;
   let listener = TcpListener::try_from(std_listener)?;
-  let rid = state.resource_table_2.add(listener);
+  let rid = state.resource_table.add(listener);
   Ok(serde_json::json!({ "rid": rid }))
 }
 
@@ -152,7 +152,7 @@ fn op_close(
     .unwrap();
   debug!("close rid={}", rid);
   state
-    .resource_table_2
+    .resource_table
     .close(rid)
     .map(|_| serde_json::json!(()))
     .ok_or_else(bad_resource_id)
@@ -174,11 +174,11 @@ async fn op_accept(
 
   let listener = state
     .borrow()
-    .resource_table_2
+    .resource_table
     .get::<TcpListener>(rid)
     .ok_or_else(bad_resource_id)?;
   let stream = listener.accept().await?;
-  let rid = state.borrow_mut().resource_table_2.add(stream);
+  let rid = state.borrow_mut().resource_table.add(stream);
   Ok(serde_json::json!({ "rid": rid }))
 }
 
@@ -199,7 +199,7 @@ async fn op_read(
 
   let stream = state
     .borrow()
-    .resource_table_2
+    .resource_table
     .get::<TcpStream>(rid)
     .ok_or_else(bad_resource_id)?;
   let nread = stream.read(&mut bufs[0]).await?;
@@ -223,7 +223,7 @@ async fn op_write(
 
   let stream = state
     .borrow()
-    .resource_table_2
+    .resource_table
     .get::<TcpStream>(rid)
     .ok_or_else(bad_resource_id)?;
   let nwritten = stream.write(&bufs[0]).await?;

core/lib.rs

@@ -17,7 +17,6 @@ mod normalize_path;
 mod ops;
 pub mod plugin_api;
 mod resources;
-mod resources2;
 mod runtime;
 mod shared_queue;
 mod zero_copy_buf;
@@ -64,10 +63,9 @@ pub use crate::ops::OpFn;
 pub use crate::ops::OpId;
 pub use crate::ops::OpState;
 pub use crate::ops::OpTable;
+pub use crate::resources::Resource;
+pub use crate::resources::ResourceId;
 pub use crate::resources::ResourceTable;
-pub use crate::resources2::Resource;
-pub use crate::resources2::ResourceId;
-pub use crate::resources2::ResourceTable2;
 pub use crate::runtime::GetErrorClassFn;
 pub use crate::runtime::JsErrorCreateFn;
 pub use crate::runtime::JsRuntime;

core/ops.rs

@@ -4,6 +4,8 @@ use crate::error::bad_resource_id;
 use crate::error::type_error;
 use crate::error::AnyError;
 use crate::gotham_state::GothamState;
+use crate::resources::ResourceTable;
+use crate::runtime::GetErrorClassFn;
 use crate::BufVec;
 use crate::ZeroCopyBuf;
 use futures::Future;
@@ -33,10 +35,9 @@ pub enum Op {
 
 /// Maintains the resources and ops inside a JS runtime.
 pub struct OpState {
-  pub resource_table: crate::ResourceTable,
-  pub resource_table_2: crate::resources2::ResourceTable,
+  pub resource_table: ResourceTable,
   pub op_table: OpTable,
-  pub get_error_class_fn: crate::runtime::GetErrorClassFn,
+  pub get_error_class_fn: GetErrorClassFn,
   gotham_state: GothamState,
 }
 
@@ -47,7 +48,6 @@ impl Default for OpState {
   fn default() -> OpState {
     OpState {
       resource_table: Default::default(),
-      resource_table_2: Default::default(),
       op_table: OpTable::default(),
       get_error_class_fn: &|_| "Error",
       gotham_state: Default::default(),
@@ -279,7 +279,11 @@ pub fn op_resources(
   _args: Value,
   _zero_copy: &mut [ZeroCopyBuf],
 ) -> Result<Value, AnyError> {
-  let serialized_resources = state.resource_table.entries();
+  let serialized_resources: HashMap<u32, String> = state
+    .resource_table
+    .names()
+    .map(|(rid, name)| (rid, name.to_string()))
+    .collect();
   Ok(json!(serialized_resources))
 }
 
@@ -300,5 +304,6 @@ pub fn op_close(
     .resource_table
     .close(rid as u32)
     .ok_or_else(bad_resource_id)?;
+
   Ok(json!({}))
 }