refactor(core): bake single-thread assumptions into spawn/spawn_blocking (#19056)

Partially supersedes #19016.

This migrates `spawn` and `spawn_blocking` to `deno_core`, and removes
the requirement for `spawn` tasks to be `Send` given our single-threaded
executor.

While we don't need to technically do anything w/`spawn_blocking`, this
allows us to have a single `JoinHandle` type that works for both cases,
and allows us to more easily experiment with alternative
`spawn_blocking` implementations that do not require tokio (ie: rayon).

Async ops (+~35%):

Before: 

```
time 1310 ms rate 763358
time 1267 ms rate 789265
time 1259 ms rate 794281
time 1266 ms rate 789889
```

After:

```
time 956 ms rate 1046025
time 954 ms rate 1048218
time 924 ms rate 1082251
time 920 ms rate 1086956
```

HTTP serve (+~4.4%):

Before:

```
Running 10s test @ http://localhost:4500
  2 threads and 10 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    68.78us   19.77us   1.43ms   86.84%
    Req/Sec    68.78k     5.00k   73.84k    91.58%
  1381833 requests in 10.10s, 167.36MB read
Requests/sec: 136823.29
Transfer/sec:     16.57MB
```

After:

```
Running 10s test @ http://localhost:4500
  2 threads and 10 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    63.12us   17.43us   1.11ms   85.13%
    Req/Sec    71.82k     3.71k   77.02k    79.21%
  1443195 requests in 10.10s, 174.79MB read
Requests/sec: 142921.99
Transfer/sec:     17.31MB
```

Suggested-By: alice@ryhl.io
Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

1 files changed, 17 insertions, 31 deletions

diff --git a/ext/node/ops/crypto/mod.rs b/ext/node/ops/crypto/mod.rs
index 9e1a3da98..0f8feb2a9 100644
--- a/ext/node/ops/crypto/mod.rs
+++ b/ext/node/ops/crypto/mod.rs
@@ -4,6 +4,7 @@ use deno_core::error::type_error;
 use deno_core::error::AnyError;
 use deno_core::op;
 use deno_core::serde_v8;
+use deno_core::task::spawn_blocking;
 use deno_core::OpState;
 use deno_core::ResourceId;
 use deno_core::StringOrBuffer;
@@ -57,12 +58,7 @@ pub async fn op_node_check_prime_async(
   checks: usize,
 ) -> Result<bool, AnyError> {
   // TODO(@littledivy): use rayon for CPU-bound tasks
-  Ok(
-    tokio::task::spawn_blocking(move || {
-      primes::is_probably_prime(&num, checks)
-    })
-    .await?,
-  )
+  Ok(spawn_blocking(move || primes::is_probably_prime(&num, checks)).await?)
 }
 
 #[op]
@@ -74,10 +70,8 @@ pub fn op_node_check_prime_bytes_async(
   // TODO(@littledivy): use rayon for CPU-bound tasks
   Ok(async move {
     Ok(
-      tokio::task::spawn_blocking(move || {
-        primes::is_probably_prime(&candidate, checks)
-      })
-      .await?,
+      spawn_blocking(move || primes::is_probably_prime(&candidate, checks))
+        .await?,
     )
   })
 }
@@ -462,7 +456,7 @@ pub async fn op_node_pbkdf2_async(
   digest: String,
   keylen: usize,
 ) -> Result<ZeroCopyBuf, AnyError> {
-  tokio::task::spawn_blocking(move || {
+  spawn_blocking(move || {
     let mut derived_key = vec![0; keylen];
     pbkdf2_sync(&password, &salt, iterations, &digest, &mut derived_key)
       .map(|_| derived_key.into())
@@ -477,7 +471,7 @@ pub fn op_node_generate_secret(buf: &mut [u8]) {
 
 #[op]
 pub async fn op_node_generate_secret_async(len: i32) -> ZeroCopyBuf {
-  tokio::task::spawn_blocking(move || {
+  spawn_blocking(move || {
     let mut buf = vec![0u8; len as usize];
     rand::thread_rng().fill(&mut buf[..]);
     buf.into()
@@ -535,7 +529,7 @@ pub async fn op_node_hkdf_async(
   info: ZeroCopyBuf,
   okm_len: usize,
 ) -> Result<ZeroCopyBuf, AnyError> {
-  tokio::task::spawn_blocking(move || {
+  spawn_blocking(move || {
     let mut okm = vec![0u8; okm_len];
     hkdf_sync(&hash, &ikm, &salt, &info, &mut okm)?;
     Ok(okm.into())
@@ -578,10 +572,7 @@ pub async fn op_node_generate_rsa_async(
   modulus_length: usize,
   public_exponent: usize,
 ) -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
-  tokio::task::spawn_blocking(move || {
-    generate_rsa(modulus_length, public_exponent)
-  })
-  .await?
+  spawn_blocking(move || generate_rsa(modulus_length, public_exponent)).await?
 }
 
 fn dsa_generate(
@@ -635,10 +626,7 @@ pub async fn op_node_dsa_generate_async(
   modulus_length: usize,
   divisor_length: usize,
 ) -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
-  tokio::task::spawn_blocking(move || {
-    dsa_generate(modulus_length, divisor_length)
-  })
-  .await?
+  spawn_blocking(move || dsa_generate(modulus_length, divisor_length)).await?
 }
 
 fn ec_generate(
@@ -677,7 +665,7 @@ pub fn op_node_ec_generate(
 pub async fn op_node_ec_generate_async(
   named_curve: String,
 ) -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
-  tokio::task::spawn_blocking(move || ec_generate(&named_curve)).await?
+  spawn_blocking(move || ec_generate(&named_curve)).await?
 }
 
 fn ed25519_generate() -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
@@ -704,7 +692,7 @@ pub fn op_node_ed25519_generate() -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError
 #[op]
 pub async fn op_node_ed25519_generate_async(
 ) -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
-  tokio::task::spawn_blocking(ed25519_generate).await?
+  spawn_blocking(ed25519_generate).await?
 }
 
 fn x25519_generate() -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
@@ -739,7 +727,7 @@ pub fn op_node_x25519_generate() -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError>
 #[op]
 pub async fn op_node_x25519_generate_async(
 ) -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
-  tokio::task::spawn_blocking(x25519_generate).await?
+  spawn_blocking(x25519_generate).await?
 }
 
 fn dh_generate_group(
@@ -772,7 +760,7 @@ pub fn op_node_dh_generate_group(
 pub async fn op_node_dh_generate_group_async(
   group_name: String,
 ) -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
-  tokio::task::spawn_blocking(move || dh_generate_group(&group_name)).await?
+  spawn_blocking(move || dh_generate_group(&group_name)).await?
 }
 
 fn dh_generate(
@@ -806,10 +794,8 @@ pub async fn op_node_dh_generate_async(
   prime_len: usize,
   generator: usize,
 ) -> Result<(ZeroCopyBuf, ZeroCopyBuf), AnyError> {
-  tokio::task::spawn_blocking(move || {
-    dh_generate(prime.as_deref(), prime_len, generator)
-  })
-  .await?
+  spawn_blocking(move || dh_generate(prime.as_deref(), prime_len, generator))
+    .await?
 }
 
 #[op]
@@ -885,7 +871,7 @@ pub async fn op_node_scrypt_async(
   parallelization: u32,
   maxmem: u32,
 ) -> Result<ZeroCopyBuf, AnyError> {
-  tokio::task::spawn_blocking(move || {
+  spawn_blocking(move || {
     let mut output_buffer = vec![0u8; keylen as usize];
     let res = scrypt(
       password,
@@ -1081,5 +1067,5 @@ pub fn op_node_gen_prime(size: usize) -> ZeroCopyBuf {
 pub async fn op_node_gen_prime_async(
   size: usize,
 ) -> Result<ZeroCopyBuf, AnyError> {
-  Ok(tokio::task::spawn_blocking(move || gen_prime(size)).await?)
+  Ok(spawn_blocking(move || gen_prime(size)).await?)
 }