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, 7 insertions, 6 deletions

diff --git a/ext/cache/sqlite.rs b/ext/cache/sqlite.rs
index 2853f793d..4eb9924c7 100644
--- a/ext/cache/sqlite.rs
+++ b/ext/cache/sqlite.rs
@@ -10,6 +10,7 @@ use std::time::UNIX_EPOCH;
 use async_trait::async_trait;
 use deno_core::error::AnyError;
 use deno_core::parking_lot::Mutex;
+use deno_core::task::spawn_blocking;
 use deno_core::AsyncRefCell;
 use deno_core::AsyncResult;
 use deno_core::ByteString;
@@ -99,7 +100,7 @@ impl Cache for SqliteBackedCache {
   async fn storage_open(&self, cache_name: String) -> Result<i64, AnyError> {
     let db = self.connection.clone();
     let cache_storage_dir = self.cache_storage_dir.clone();
-    tokio::task::spawn_blocking(move || {
+    spawn_blocking(move || {
       let db = db.lock();
       db.execute(
         "INSERT OR IGNORE INTO cache_storage (cache_name) VALUES (?1)",
@@ -124,7 +125,7 @@ impl Cache for SqliteBackedCache {
   /// Note: this doesn't check the disk, it only checks the sqlite db.
   async fn storage_has(&self, cache_name: String) -> Result<bool, AnyError> {
     let db = self.connection.clone();
-    tokio::task::spawn_blocking(move || {
+    spawn_blocking(move || {
       let db = db.lock();
       let cache_exists = db.query_row(
         "SELECT count(id) FROM cache_storage WHERE cache_name = ?1",
@@ -143,7 +144,7 @@ impl Cache for SqliteBackedCache {
   async fn storage_delete(&self, cache_name: String) -> Result<bool, AnyError> {
     let db = self.connection.clone();
     let cache_storage_dir = self.cache_storage_dir.clone();
-    tokio::task::spawn_blocking(move || {
+    spawn_blocking(move || {
       let db = db.lock();
       let maybe_cache_id = db
         .query_row(
@@ -210,7 +211,7 @@ impl Cache for SqliteBackedCache {
   > {
     let db = self.connection.clone();
     let cache_storage_dir = self.cache_storage_dir.clone();
-    let query_result = tokio::task::spawn_blocking(move || {
+    let query_result = spawn_blocking(move || {
       let db = db.lock();
       let result = db.query_row(
         "SELECT response_body_key, response_headers, response_status, response_status_text, request_headers
@@ -269,7 +270,7 @@ impl Cache for SqliteBackedCache {
     request: CacheDeleteRequest,
   ) -> Result<bool, AnyError> {
     let db = self.connection.clone();
-    tokio::task::spawn_blocking(move || {
+    spawn_blocking(move || {
       // TODO(@satyarohith): remove the response body from disk if one exists
       let db = db.lock();
       let rows_effected = db.execute(
@@ -287,7 +288,7 @@ async fn insert_cache_asset(
   put: CachePutRequest,
   response_body_key: Option<String>,
 ) -> Result<Option<String>, deno_core::anyhow::Error> {
-  tokio::task::spawn_blocking(move || {
+  spawn_blocking(move || {
     let maybe_response_body = {
       let db = db.lock();
       db.query_row(