diff options
author | Matt Mastracci <matthew@mastracci.com> | 2023-05-14 15:40:01 -0600 |
---|---|---|
committer | GitHub <noreply@github.com> | 2023-05-14 15:40:01 -0600 |
commit | 9845361153f35f6a68a82eb3a13845fddbeab026 (patch) | |
tree | 307b58f09cac9e681255dac74f487f8da70d76d2 /cli | |
parent | b99159bf14d15418a7dbb22e9ce78b15d52971cc (diff) |
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>
Diffstat (limited to 'cli')
-rw-r--r-- | cli/cache/cache_db.rs | 5 | ||||
-rw-r--r-- | cli/cache/incremental.rs | 5 | ||||
-rw-r--r-- | cli/lsp/client.rs | 7 | ||||
-rw-r--r-- | cli/lsp/diagnostics.rs | 14 | ||||
-rw-r--r-- | cli/lsp/language_server.rs | 3 | ||||
-rw-r--r-- | cli/lsp/parent_process_checker.rs | 3 | ||||
-rw-r--r-- | cli/lsp/testing/execution.rs | 14 | ||||
-rw-r--r-- | cli/main.rs | 4 | ||||
-rw-r--r-- | cli/npm/resolvers/common.rs | 3 | ||||
-rw-r--r-- | cli/npm/resolvers/local.rs | 5 | ||||
-rw-r--r-- | cli/tests/integration/cert_tests.rs | 197 | ||||
-rw-r--r-- | cli/tests/integration/inspector_tests.rs | 2 | ||||
-rw-r--r-- | cli/tests/integration/run_tests.rs | 77 | ||||
-rw-r--r-- | cli/tools/bench.rs | 14 | ||||
-rw-r--r-- | cli/tools/fmt.rs | 3 | ||||
-rw-r--r-- | cli/tools/repl/mod.rs | 3 | ||||
-rw-r--r-- | cli/tools/task.rs | 14 | ||||
-rw-r--r-- | cli/tools/test.rs | 24 | ||||
-rw-r--r-- | cli/tools/upgrade.rs | 3 | ||||
-rw-r--r-- | cli/util/draw_thread.rs | 3 | ||||
-rw-r--r-- | cli/util/fs.rs | 3 |
21 files changed, 204 insertions, 202 deletions
diff --git a/cli/cache/cache_db.rs b/cli/cache/cache_db.rs index 90840de1a..e05ecd962 100644 --- a/cli/cache/cache_db.rs +++ b/cli/cache/cache_db.rs @@ -3,6 +3,7 @@ use deno_core::error::AnyError; use deno_core::parking_lot::Mutex; use deno_core::parking_lot::MutexGuard; +use deno_core::task::spawn_blocking; use deno_runtime::deno_webstorage::rusqlite; use deno_runtime::deno_webstorage::rusqlite::Connection; use deno_runtime::deno_webstorage::rusqlite::OptionalExtension; @@ -95,7 +96,7 @@ impl Drop for CacheDB { // Hand off SQLite connection to another thread to do the surprisingly expensive cleanup let inner = inner.into_inner().into_inner(); if let Some(conn) = inner { - tokio::task::spawn_blocking(move || { + spawn_blocking(move || { drop(conn); log::trace!( "Cleaned up SQLite connection at {}", @@ -168,7 +169,7 @@ impl CacheDB { fn spawn_eager_init_thread(&self) { let clone = self.clone(); debug_assert!(tokio::runtime::Handle::try_current().is_ok()); - tokio::task::spawn_blocking(move || { + spawn_blocking(move || { let lock = clone.conn.lock(); clone.initialize(&lock); }); diff --git a/cli/cache/incremental.rs b/cli/cache/incremental.rs index deb30cdd1..c50b876fa 100644 --- a/cli/cache/incremental.rs +++ b/cli/cache/incremental.rs @@ -7,9 +7,10 @@ use std::path::PathBuf; use deno_core::error::AnyError; use deno_core::parking_lot::Mutex; use deno_core::serde_json; +use deno_core::task::spawn; +use deno_core::task::JoinHandle; use deno_runtime::deno_webstorage::rusqlite::params; use serde::Serialize; -use tokio::task::JoinHandle; use super::cache_db::CacheDB; use super::cache_db::CacheDBConfiguration; @@ -93,7 +94,7 @@ impl IncrementalCacheInner { tokio::sync::mpsc::unbounded_channel::<ReceiverMessage>(); // sqlite isn't `Sync`, so we do all the updating on a dedicated task - let handle = tokio::task::spawn(async move { + let handle = spawn(async move { while let Some(message) = receiver.recv().await { match message { ReceiverMessage::Update(path, hash) => { diff --git a/cli/lsp/client.rs b/cli/lsp/client.rs index d24d4c2a9..4923a4585 100644 --- a/cli/lsp/client.rs +++ b/cli/lsp/client.rs @@ -8,6 +8,7 @@ use deno_core::anyhow::bail; use deno_core::error::AnyError; use deno_core::serde_json; use deno_core::serde_json::Value; +use deno_core::task::spawn; use tower_lsp::lsp_types as lsp; use tower_lsp::lsp_types::ConfigurationItem; @@ -56,7 +57,7 @@ impl Client { ) { // do on a task in case the caller currently is in the lsp lock let client = self.0.clone(); - tokio::task::spawn(async move { + spawn(async move { client.send_registry_state_notification(params).await; }); } @@ -64,7 +65,7 @@ impl Client { pub fn send_test_notification(&self, params: TestingNotification) { // do on a task in case the caller currently is in the lsp lock let client = self.0.clone(); - tokio::task::spawn(async move { + spawn(async move { client.send_test_notification(params).await; }); } @@ -77,7 +78,7 @@ impl Client { // do on a task in case the caller currently is in the lsp lock let client = self.0.clone(); let message = message.to_string(); - tokio::task::spawn(async move { + spawn(async move { client.show_message(message_type, message).await; }); } diff --git a/cli/lsp/diagnostics.rs b/cli/lsp/diagnostics.rs index 0f96a498b..7b5a30a0e 100644 --- a/cli/lsp/diagnostics.rs +++ b/cli/lsp/diagnostics.rs @@ -25,6 +25,8 @@ use deno_core::resolve_url; use deno_core::serde::Deserialize; use deno_core::serde_json; use deno_core::serde_json::json; +use deno_core::task::spawn; +use deno_core::task::JoinHandle; use deno_core::ModuleSpecifier; use deno_graph::Resolution; use deno_graph::ResolutionError; @@ -197,9 +199,9 @@ impl DiagnosticsServer { runtime.block_on(async { let mut token = CancellationToken::new(); - let mut ts_handle: Option<tokio::task::JoinHandle<()>> = None; - let mut lint_handle: Option<tokio::task::JoinHandle<()>> = None; - let mut deps_handle: Option<tokio::task::JoinHandle<()>> = None; + let mut ts_handle: Option<JoinHandle<()>> = None; + let mut lint_handle: Option<JoinHandle<()>> = None; + let mut deps_handle: Option<JoinHandle<()>> = None; let diagnostics_publisher = DiagnosticsPublisher::new(client.clone()); loop { @@ -213,7 +215,7 @@ impl DiagnosticsServer { diagnostics_publisher.clear().await; let previous_ts_handle = ts_handle.take(); - ts_handle = Some(tokio::spawn({ + ts_handle = Some(spawn({ let performance = performance.clone(); let diagnostics_publisher = diagnostics_publisher.clone(); let ts_server = ts_server.clone(); @@ -265,7 +267,7 @@ impl DiagnosticsServer { })); let previous_deps_handle = deps_handle.take(); - deps_handle = Some(tokio::spawn({ + deps_handle = Some(spawn({ let performance = performance.clone(); let diagnostics_publisher = diagnostics_publisher.clone(); let token = token.clone(); @@ -293,7 +295,7 @@ impl DiagnosticsServer { })); let previous_lint_handle = lint_handle.take(); - lint_handle = Some(tokio::spawn({ + lint_handle = Some(spawn({ let performance = performance.clone(); let diagnostics_publisher = diagnostics_publisher.clone(); let token = token.clone(); diff --git a/cli/lsp/language_server.rs b/cli/lsp/language_server.rs index de5cd6f09..e76ea0040 100644 --- a/cli/lsp/language_server.rs +++ b/cli/lsp/language_server.rs @@ -8,6 +8,7 @@ use deno_core::resolve_url; use deno_core::serde_json; use deno_core::serde_json::json; use deno_core::serde_json::Value; +use deno_core::task::spawn; use deno_core::ModuleSpecifier; use deno_runtime::deno_fs; use deno_runtime::deno_node::NodeResolver; @@ -240,7 +241,7 @@ impl LanguageServer { let cli_options = result.cli_options; let roots = result.roots; let open_docs = result.open_docs; - let handle = tokio::task::spawn_local(async move { + let handle = spawn(async move { create_graph_for_caching(cli_options, roots, open_docs).await }); if let Err(err) = handle.await.unwrap() { diff --git a/cli/lsp/parent_process_checker.rs b/cli/lsp/parent_process_checker.rs index 4cc3bcef3..f83543c04 100644 --- a/cli/lsp/parent_process_checker.rs +++ b/cli/lsp/parent_process_checker.rs @@ -1,5 +1,6 @@ // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +use deno_core::task::spawn; use tokio::time::sleep; use tokio::time::Duration; @@ -7,7 +8,7 @@ use tokio::time::Duration; /// provided process id. Once that process no longer exists /// it will terminate the current process. pub fn start(parent_process_id: u32) { - tokio::task::spawn(async move { + spawn(async move { loop { sleep(Duration::from_secs(30)).await; diff --git a/cli/lsp/testing/execution.rs b/cli/lsp/testing/execution.rs index 4834cd0c9..ce8c8b5ac 100644 --- a/cli/lsp/testing/execution.rs +++ b/cli/lsp/testing/execution.rs @@ -24,9 +24,11 @@ use deno_core::futures::stream; use deno_core::futures::StreamExt; use deno_core::parking_lot::Mutex; use deno_core::parking_lot::RwLock; +use deno_core::task::spawn; +use deno_core::task::spawn_blocking; use deno_core::ModuleSpecifier; use deno_runtime::permissions::Permissions; -use deno_runtime::tokio_util::run_local; +use deno_runtime::tokio_util::create_and_run_current_thread; use indexmap::IndexMap; use std::collections::HashMap; use std::collections::HashSet; @@ -284,7 +286,7 @@ impl TestRun { }; let token = self.token.clone(); - tokio::task::spawn_blocking(move || { + spawn_blocking(move || { if fail_fast_tracker.should_stop() { return Ok(()); } @@ -292,13 +294,13 @@ impl TestRun { let file_result = if token.is_cancelled() { Ok(()) } else { - run_local(test::test_specifier( - &worker_factory, + create_and_run_current_thread(test::test_specifier( + worker_factory, permissions, specifier, sender.clone(), fail_fast_tracker, - &test::TestSpecifierOptions { + test::TestSpecifierOptions { filter, shuffle: None, trace_ops: false, @@ -331,7 +333,7 @@ impl TestRun { )); let handler = { - tokio::task::spawn(async move { + spawn(async move { let earlier = Instant::now(); let mut summary = test::TestSummary::new(); let mut used_only = false; diff --git a/cli/main.rs b/cli/main.rs index 75425cf10..023d5a208 100644 --- a/cli/main.rs +++ b/cli/main.rs @@ -46,7 +46,7 @@ use deno_core::error::AnyError; use deno_core::error::JsError; use deno_runtime::colors; use deno_runtime::fmt_errors::format_js_error; -use deno_runtime::tokio_util::run_local; +use deno_runtime::tokio_util::create_and_run_current_thread; use factory::CliFactory; use std::env; use std::env::current_exe; @@ -294,7 +294,7 @@ pub fn main() { run_subcommand(flags).await }; - let exit_code = unwrap_or_exit(run_local(future)); + let exit_code = unwrap_or_exit(create_and_run_current_thread(future)); std::process::exit(exit_code); } diff --git a/cli/npm/resolvers/common.rs b/cli/npm/resolvers/common.rs index fc040a7cc..c91b206cf 100644 --- a/cli/npm/resolvers/common.rs +++ b/cli/npm/resolvers/common.rs @@ -9,6 +9,7 @@ use async_trait::async_trait; use deno_ast::ModuleSpecifier; use deno_core::error::AnyError; use deno_core::futures; +use deno_core::task::spawn; use deno_core::url::Url; use deno_npm::NpmPackageId; use deno_npm::NpmResolutionPackage; @@ -71,7 +72,7 @@ pub async fn cache_packages( assert_eq!(package.copy_index, 0); // the caller should not provide any of these let cache = cache.clone(); let registry_url = registry_url.clone(); - let handle = tokio::task::spawn(async move { + let handle = spawn(async move { cache .ensure_package(&package.pkg_id.nv, &package.dist, ®istry_url) .await diff --git a/cli/npm/resolvers/local.rs b/cli/npm/resolvers/local.rs index cd1dc3671..b2ad08357 100644 --- a/cli/npm/resolvers/local.rs +++ b/cli/npm/resolvers/local.rs @@ -18,6 +18,8 @@ use deno_ast::ModuleSpecifier; use deno_core::anyhow::bail; use deno_core::anyhow::Context; use deno_core::error::AnyError; +use deno_core::task::spawn; +use deno_core::task::JoinHandle; use deno_core::url::Url; use deno_npm::resolution::NpmResolutionSnapshot; use deno_npm::NpmPackageCacheFolderId; @@ -27,7 +29,6 @@ use deno_runtime::deno_fs; use deno_runtime::deno_node::NodePermissions; use deno_runtime::deno_node::NodeResolutionMode; use deno_runtime::deno_node::PackageJson; -use tokio::task::JoinHandle; use crate::npm::cache::mixed_case_package_name_encode; use crate::npm::cache::should_sync_download; @@ -277,7 +278,7 @@ async fn sync_resolution_with_fs( let cache = cache.clone(); let registry_url = registry_url.clone(); let package = package.clone(); - let handle = tokio::task::spawn(async move { + let handle = spawn(async move { cache .ensure_package(&package.pkg_id.nv, &package.dist, ®istry_url) .await?; diff --git a/cli/tests/integration/cert_tests.rs b/cli/tests/integration/cert_tests.rs index d3da6d75a..b04f2d35e 100644 --- a/cli/tests/integration/cert_tests.rs +++ b/cli/tests/integration/cert_tests.rs @@ -11,7 +11,6 @@ use std::process::Command; use std::sync::Arc; use test_util as util; use test_util::TempDir; -use tokio::task::LocalSet; use util::TestContext; itest_flaky!(cafile_url_imports { @@ -219,113 +218,99 @@ fn cafile_bundle_remote_exports() { #[tokio::test] async fn listen_tls_alpn() { - // TLS streams require the presence of an ambient local task set to gracefully - // close dropped connections in the background. - LocalSet::new() - .run_until(async { - let mut child = util::deno_cmd() - .current_dir(util::testdata_path()) - .arg("run") - .arg("--unstable") - .arg("--quiet") - .arg("--allow-net") - .arg("--allow-read") - .arg("./cert/listen_tls_alpn.ts") - .arg("4504") - .stdout(std::process::Stdio::piped()) - .spawn() - .unwrap(); - let stdout = child.stdout.as_mut().unwrap(); - let mut msg = [0; 5]; - let read = stdout.read(&mut msg).unwrap(); - assert_eq!(read, 5); - assert_eq!(&msg, b"READY"); - - let mut reader = &mut BufReader::new(Cursor::new(include_bytes!( - "../testdata/tls/RootCA.crt" - ))); - let certs = rustls_pemfile::certs(&mut reader).unwrap(); - let mut root_store = rustls::RootCertStore::empty(); - root_store.add_parsable_certificates(&certs); - let mut cfg = rustls::ClientConfig::builder() - .with_safe_defaults() - .with_root_certificates(root_store) - .with_no_client_auth(); - cfg.alpn_protocols.push(b"foobar".to_vec()); - let cfg = Arc::new(cfg); - - let hostname = rustls::ServerName::try_from("localhost").unwrap(); - - let tcp_stream = tokio::net::TcpStream::connect("localhost:4504") - .await - .unwrap(); - let mut tls_stream = - TlsStream::new_client_side(tcp_stream, cfg, hostname); - - tls_stream.handshake().await.unwrap(); - - let (_, rustls_connection) = tls_stream.get_ref(); - let alpn = rustls_connection.alpn_protocol().unwrap(); - assert_eq!(alpn, b"foobar"); - - let status = child.wait().unwrap(); - assert!(status.success()); - }) - .await; + let mut child = util::deno_cmd() + .current_dir(util::testdata_path()) + .arg("run") + .arg("--unstable") + .arg("--quiet") + .arg("--allow-net") + .arg("--allow-read") + .arg("./cert/listen_tls_alpn.ts") + .arg("4504") + .stdout(std::process::Stdio::piped()) + .spawn() + .unwrap(); + let stdout = child.stdout.as_mut().unwrap(); + let mut msg = [0; 5]; + let read = stdout.read(&mut msg).unwrap(); + assert_eq!(read, 5); + assert_eq!(&msg, b"READY"); + + let mut reader = &mut BufReader::new(Cursor::new(include_bytes!( + "../testdata/tls/RootCA.crt" + ))); + let certs = rustls_pemfile::certs(&mut reader).unwrap(); + let mut root_store = rustls::RootCertStore::empty(); + root_store.add_parsable_certificates(&certs); + let mut cfg = rustls::ClientConfig::builder() + .with_safe_defaults() + .with_root_certificates(root_store) + .with_no_client_auth(); + cfg.alpn_protocols.push(b"foobar".to_vec()); + let cfg = Arc::new(cfg); + + let hostname = rustls::ServerName::try_from("localhost").unwrap(); + + let tcp_stream = tokio::net::TcpStream::connect("localhost:4504") + .await + .unwrap(); + let mut tls_stream = TlsStream::new_client_side(tcp_stream, cfg, hostname); + + tls_stream.handshake().await.unwrap(); + + let (_, rustls_connection) = tls_stream.get_ref(); + let alpn = rustls_connection.alpn_protocol().unwrap(); + assert_eq!(alpn, b"foobar"); + + let status = child.wait().unwrap(); + assert!(status.success()); } #[tokio::test] async fn listen_tls_alpn_fail() { - // TLS streams require the presence of an ambient local task set to gracefully - // close dropped connections in the background. - LocalSet::new() - .run_until(async { - let mut child = util::deno_cmd() - .current_dir(util::testdata_path()) - .arg("run") - .arg("--unstable") - .arg("--quiet") - .arg("--allow-net") - .arg("--allow-read") - .arg("./cert/listen_tls_alpn_fail.ts") - .arg("4505") - .stdout(std::process::Stdio::piped()) - .spawn() - .unwrap(); - let stdout = child.stdout.as_mut().unwrap(); - let mut msg = [0; 5]; - let read = stdout.read(&mut msg).unwrap(); - assert_eq!(read, 5); - assert_eq!(&msg, b"READY"); - - let mut reader = &mut BufReader::new(Cursor::new(include_bytes!( - "../testdata/tls/RootCA.crt" - ))); - let certs = rustls_pemfile::certs(&mut reader).unwrap(); - let mut root_store = rustls::RootCertStore::empty(); - root_store.add_parsable_certificates(&certs); - let mut cfg = rustls::ClientConfig::builder() - .with_safe_defaults() - .with_root_certificates(root_store) - .with_no_client_auth(); - cfg.alpn_protocols.push(b"boofar".to_vec()); - let cfg = Arc::new(cfg); - - let hostname = rustls::ServerName::try_from("localhost").unwrap(); - - let tcp_stream = tokio::net::TcpStream::connect("localhost:4505") - .await - .unwrap(); - let mut tls_stream = - TlsStream::new_client_side(tcp_stream, cfg, hostname); - - tls_stream.handshake().await.unwrap_err(); - - let (_, rustls_connection) = tls_stream.get_ref(); - assert!(rustls_connection.alpn_protocol().is_none()); - - let status = child.wait().unwrap(); - assert!(status.success()); - }) - .await; + let mut child = util::deno_cmd() + .current_dir(util::testdata_path()) + .arg("run") + .arg("--unstable") + .arg("--quiet") + .arg("--allow-net") + .arg("--allow-read") + .arg("./cert/listen_tls_alpn_fail.ts") + .arg("4505") + .stdout(std::process::Stdio::piped()) + .spawn() + .unwrap(); + let stdout = child.stdout.as_mut().unwrap(); + let mut msg = [0; 5]; + let read = stdout.read(&mut msg).unwrap(); + assert_eq!(read, 5); + assert_eq!(&msg, b"READY"); + + let mut reader = &mut BufReader::new(Cursor::new(include_bytes!( + "../testdata/tls/RootCA.crt" + ))); + let certs = rustls_pemfile::certs(&mut reader).unwrap(); + let mut root_store = rustls::RootCertStore::empty(); + root_store.add_parsable_certificates(&certs); + let mut cfg = rustls::ClientConfig::builder() + .with_safe_defaults() + .with_root_certificates(root_store) + .with_no_client_auth(); + cfg.alpn_protocols.push(b"boofar".to_vec()); + let cfg = Arc::new(cfg); + + let hostname = rustls::ServerName::try_from("localhost").unwrap(); + + let tcp_stream = tokio::net::TcpStream::connect("localhost:4505") + .await + .unwrap(); + let mut tls_stream = TlsStream::new_client_side(tcp_stream, cfg, hostname); + + tls_stream.handshake().await.unwrap_err(); + + let (_, rustls_connection) = tls_stream.get_ref(); + assert!(rustls_connection.alpn_protocol().is_none()); + + let status = child.wait().unwrap(); + assert!(status.success()); } diff --git a/cli/tests/integration/inspector_tests.rs b/cli/tests/integration/inspector_tests.rs index cf66c4adc..8fa9ec85c 100644 --- a/cli/tests/integration/inspector_tests.rs +++ b/cli/tests/integration/inspector_tests.rs @@ -29,7 +29,7 @@ where Fut::Output: Send + 'static, { fn execute(&self, fut: Fut) { - tokio::task::spawn(fut); + deno_core::task::spawn(fut); } } diff --git a/cli/tests/integration/run_tests.rs b/cli/tests/integration/run_tests.rs index e6ea85da4..bc717351a 100644 --- a/cli/tests/integration/run_tests.rs +++ b/cli/tests/integration/run_tests.rs @@ -9,7 +9,6 @@ use std::process::Stdio; use std::time::Duration; use test_util as util; use test_util::TempDir; -use tokio::task::LocalSet; use trust_dns_client::serialize::txt::Lexer; use trust_dns_client::serialize::txt::Parser; use util::assert_contains; @@ -3886,50 +3885,44 @@ async fn test_resolve_dns() { #[tokio::test] async fn http2_request_url() { - // TLS streams require the presence of an ambient local task set to gracefully - // close dropped connections in the background. - LocalSet::new() - .run_until(async { - let mut child = util::deno_cmd() - .current_dir(util::testdata_path()) - .arg("run") - .arg("--unstable") - .arg("--quiet") - .arg("--allow-net") - .arg("--allow-read") - .arg("./run/http2_request_url.ts") - .arg("4506") - .stdout(std::process::Stdio::piped()) - .spawn() - .unwrap(); - let stdout = child.stdout.as_mut().unwrap(); - let mut buffer = [0; 5]; - let read = stdout.read(&mut buffer).unwrap(); - assert_eq!(read, 5); - let msg = std::str::from_utf8(&buffer).unwrap(); - assert_eq!(msg, "READY"); - - let cert = reqwest::Certificate::from_pem(include_bytes!( - "../testdata/tls/RootCA.crt" - )) - .unwrap(); + let mut child = util::deno_cmd() + .current_dir(util::testdata_path()) + .arg("run") + .arg("--unstable") + .arg("--quiet") + .arg("--allow-net") + .arg("--allow-read") + .arg("./run/http2_request_url.ts") + .arg("4506") + .stdout(std::process::Stdio::piped()) + .spawn() + .unwrap(); + let stdout = child.stdout.as_mut().unwrap(); + let mut buffer = [0; 5]; + let read = stdout.read(&mut buffer).unwrap(); + assert_eq!(read, 5); + let msg = std::str::from_utf8(&buffer).unwrap(); + assert_eq!(msg, "READY"); - let client = reqwest::Client::builder() - .add_root_certificate(cert) - .http2_prior_knowledge() - .build() - .unwrap(); + let cert = reqwest::Certificate::from_pem(include_bytes!( + "../testdata/tls/RootCA.crt" + )) + .unwrap(); + + let client = reqwest::Client::builder() + .add_root_certificate(cert) + .http2_prior_knowledge() + .build() + .unwrap(); - let res = client.get("http://127.0.0.1:4506").send().await.unwrap(); - assert_eq!(200, res.status()); + let res = client.get("http://127.0.0.1:4506").send().await.unwrap(); + assert_eq!(200, res.status()); - let body = res.text().await.unwrap(); - assert_eq!(body, "http://127.0.0.1:4506/"); + let body = res.text().await.unwrap(); + assert_eq!(body, "http://127.0.0.1:4506/"); - child.kill().unwrap(); - child.wait().unwrap(); - }) - .await; + child.kill().unwrap(); + child.wait().unwrap(); } #[cfg(not(windows))] @@ -4173,7 +4166,7 @@ where Fut::Output: Send + 'static, { fn execute(&self, fut: Fut) { - tokio::task::spawn(fut); + deno_core::task::spawn(fut); } } diff --git a/cli/tools/bench.rs b/cli/tools/bench.rs index 3d5f99aba..107fd2b9b 100644 --- a/cli/tools/bench.rs +++ b/cli/tools/bench.rs @@ -27,11 +27,13 @@ use deno_core::futures::FutureExt; use deno_core::futures::StreamExt; use deno_core::located_script_name; use deno_core::serde_v8; +use deno_core::task::spawn; +use deno_core::task::spawn_blocking; use deno_core::v8; use deno_core::ModuleSpecifier; use deno_runtime::permissions::Permissions; use deno_runtime::permissions::PermissionsContainer; -use deno_runtime::tokio_util::run_local; +use deno_runtime::tokio_util::create_and_run_current_thread; use indexmap::IndexMap; use indexmap::IndexSet; use log::Level; @@ -436,7 +438,7 @@ async fn check_specifiers( /// Run a single specifier as an executable bench module. async fn bench_specifier( - worker_factory: &CliMainWorkerFactory, + worker_factory: Arc<CliMainWorkerFactory>, permissions: Permissions, specifier: ModuleSpecifier, sender: UnboundedSender<BenchEvent>, @@ -522,15 +524,15 @@ async fn bench_specifiers( let specifier = specifier; let sender = sender.clone(); let options = option_for_handles.clone(); - tokio::task::spawn_blocking(move || { + spawn_blocking(move || { let future = bench_specifier( - &worker_factory, + worker_factory, permissions, specifier, sender, options.filter, ); - run_local(future) + create_and_run_current_thread(future) }) }); @@ -539,7 +541,7 @@ async fn bench_specifiers( .collect::<Vec<Result<Result<(), AnyError>, tokio::task::JoinError>>>(); let handler = { - tokio::task::spawn(async move { + spawn(async move { let mut used_only = false; let mut report = BenchReport::new(); let mut reporter = diff --git a/cli/tools/fmt.rs b/cli/tools/fmt.rs index 70d2bd639..f2fec9302 100644 --- a/cli/tools/fmt.rs +++ b/cli/tools/fmt.rs @@ -28,6 +28,7 @@ use deno_core::error::generic_error; use deno_core::error::AnyError; use deno_core::futures; use deno_core::parking_lot::Mutex; +use deno_core::task::spawn_blocking; use log::debug; use log::info; use log::warn; @@ -629,7 +630,7 @@ where let handles = file_paths.iter().map(|file_path| { let f = f.clone(); let file_path = file_path.clone(); - tokio::task::spawn_blocking(move || f(file_path)) + spawn_blocking(move || f(file_path)) }); let join_results = futures::future::join_all(handles).await; diff --git a/cli/tools/repl/mod.rs b/cli/tools/repl/mod.rs index 9f4b58919..dfd9931b8 100644 --- a/cli/tools/repl/mod.rs +++ b/cli/tools/repl/mod.rs @@ -8,6 +8,7 @@ use crate::factory::CliFactory; use crate::file_fetcher::FileFetcher; use deno_core::error::AnyError; use deno_core::futures::StreamExt; +use deno_core::task::spawn_blocking; use deno_runtime::permissions::Permissions; use deno_runtime::permissions::PermissionsContainer; use rustyline::error::ReadlineError; @@ -32,7 +33,7 @@ async fn read_line_and_poll( editor: ReplEditor, ) -> Result<String, ReadlineError> { #![allow(clippy::await_holding_refcell_ref)] - let mut line_fut = tokio::task::spawn_blocking(move || editor.readline()); + let mut line_fut = spawn_blocking(move || editor.readline()); let mut poll_worker = true; let notifications_rc = repl_session.notifications.clone(); let mut notifications = notifications_rc.borrow_mut(); diff --git a/cli/tools/task.rs b/cli/tools/task.rs index bf972e2db..37a1aa1c9 100644 --- a/cli/tools/task.rs +++ b/cli/tools/task.rs @@ -21,6 +21,7 @@ use indexmap::IndexMap; use std::collections::HashMap; use std::path::PathBuf; use std::rc::Rc; +use tokio::task::LocalSet; pub async fn execute_script( flags: Flags, @@ -59,9 +60,10 @@ pub async fn execute_script( let seq_list = deno_task_shell::parser::parse(&script) .with_context(|| format!("Error parsing script '{task_name}'."))?; let env_vars = collect_env_vars(); - let exit_code = - deno_task_shell::execute(seq_list, env_vars, &cwd, Default::default()) - .await; + let local = LocalSet::new(); + let future = + deno_task_shell::execute(seq_list, env_vars, &cwd, Default::default()); + let exit_code = local.run_until(future).await; Ok(exit_code) } else if let Some(script) = package_json_scripts.get(task_name) { let package_json_deps_provider = factory.package_json_deps_provider(); @@ -109,8 +111,10 @@ pub async fn execute_script( .with_context(|| format!("Error parsing script '{task_name}'."))?; let npx_commands = resolve_npm_commands(npm_resolver, node_resolver)?; let env_vars = collect_env_vars(); - let exit_code = - deno_task_shell::execute(seq_list, env_vars, &cwd, npx_commands).await; + let local = LocalSet::new(); + let future = + deno_task_shell::execute(seq_list, env_vars, &cwd, npx_commands); + let exit_code = local.run_until(future).await; Ok(exit_code) } else { eprintln!("Task not found: {task_name}"); diff --git a/cli/tools/test.rs b/cli/tools/test.rs index 50e220a46..f78e32539 100644 --- a/cli/tools/test.rs +++ b/cli/tools/test.rs @@ -34,6 +34,8 @@ use deno_core::futures::StreamExt; use deno_core::located_script_name; use deno_core::parking_lot::Mutex; use deno_core::serde_v8; +use deno_core::task::spawn; +use deno_core::task::spawn_blocking; use deno_core::url::Url; use deno_core::v8; use deno_core::ModuleSpecifier; @@ -42,7 +44,7 @@ use deno_runtime::deno_io::StdioPipe; use deno_runtime::fmt_errors::format_js_error; use deno_runtime::permissions::Permissions; use deno_runtime::permissions::PermissionsContainer; -use deno_runtime::tokio_util::run_local; +use deno_runtime::tokio_util::create_and_run_current_thread; use indexmap::IndexMap; use indexmap::IndexSet; use log::Level; @@ -916,12 +918,12 @@ pub fn format_test_error(js_error: &JsError) -> String { /// Test a single specifier as documentation containing test programs, an executable test module or /// both. pub async fn test_specifier( - worker_factory: &CliMainWorkerFactory, + worker_factory: Arc<CliMainWorkerFactory>, permissions: Permissions, specifier: ModuleSpecifier, mut sender: TestEventSender, fail_fast_tracker: FailFastTracker, - options: &TestSpecifierOptions, + options: TestSpecifierOptions, ) -> Result<(), AnyError> { if fail_fast_tracker.should_stop() { return Ok(()); @@ -1316,7 +1318,7 @@ async fn test_specifiers( let concurrent_jobs = options.concurrent_jobs; let sender_ = sender.downgrade(); - let sigint_handler_handle = tokio::task::spawn(async move { + let sigint_handler_handle = spawn(async move { signal::ctrl_c().await.unwrap(); sender_.upgrade().map(|s| s.send(TestEvent::Sigint).ok()); }); @@ -1328,14 +1330,14 @@ async fn test_specifiers( let sender = sender.clone(); let fail_fast_tracker = FailFastTracker::new(options.fail_fast); let specifier_options = options.specifier.clone(); - tokio::task::spawn_blocking(move || { - run_local(test_specifier( - &worker_factory, + spawn_blocking(move || { + create_and_run_current_thread(test_specifier( + worker_factory, permissions, specifier, sender.clone(), fail_fast_tracker, - &specifier_options, + specifier_options, )) }) }); @@ -1350,7 +1352,7 @@ async fn test_specifiers( )); let handler = { - tokio::task::spawn(async move { + spawn(async move { let earlier = Instant::now(); let mut tests = IndexMap::new(); let mut test_steps = IndexMap::new(); @@ -1887,7 +1889,7 @@ pub async fn run_tests_with_watch( // run, a process-scoped basic exit handler is required due to a tokio // limitation where it doesn't unbind its own handler for the entire process // once a user adds one. - tokio::task::spawn(async move { + spawn(async move { loop { signal::ctrl_c().await.unwrap(); if !HAS_TEST_RUN_SIGINT_HANDLER.load(Ordering::Relaxed) { @@ -2070,7 +2072,7 @@ fn start_output_redirect_thread( sender: UnboundedSender<TestEvent>, flush_state: Arc<Mutex<Option<std::sync::mpsc::Sender<()>>>>, ) { - tokio::task::spawn_blocking(move || loop { + spawn_blocking(move || loop { let mut buffer = [0; 512]; let size = match pipe_reader.read(&mut buffer) { Ok(0) | Err(_) => break, diff --git a/cli/tools/upgrade.rs b/cli/tools/upgrade.rs index b5aefe479..cbd924755 100644 --- a/cli/tools/upgrade.rs +++ b/cli/tools/upgrade.rs @@ -17,6 +17,7 @@ use deno_core::anyhow::Context; use deno_core::error::AnyError; use deno_core::futures::future::BoxFuture; use deno_core::futures::FutureExt; +use deno_core::task::spawn; use deno_semver::Version; use once_cell::sync::Lazy; use std::borrow::Cow; @@ -198,7 +199,7 @@ pub fn check_for_upgrades( if update_checker.should_check_for_new_version() { let env = update_checker.env.clone(); // do this asynchronously on a separate task - tokio::spawn(async move { + spawn(async move { // Sleep for a small amount of time to not unnecessarily impact startup // time. tokio::time::sleep(UPGRADE_CHECK_FETCH_DELAY).await; diff --git a/cli/util/draw_thread.rs b/cli/util/draw_thread.rs index 028b20d00..2fd81a78a 100644 --- a/cli/util/draw_thread.rs +++ b/cli/util/draw_thread.rs @@ -2,6 +2,7 @@ use console_static_text::ConsoleStaticText; use deno_core::parking_lot::Mutex; +use deno_core::task::spawn_blocking; use deno_runtime::ops::tty::ConsoleSize; use once_cell::sync::Lazy; use std::sync::Arc; @@ -162,7 +163,7 @@ impl DrawThread { internal_state.has_draw_thread = true; let drawer_id = internal_state.drawer_id; - tokio::task::spawn_blocking(move || { + spawn_blocking(move || { let mut previous_size = console_size(); loop { let mut delay_ms = 120; diff --git a/cli/util/fs.rs b/cli/util/fs.rs index 94ec24fe6..658002e3b 100644 --- a/cli/util/fs.rs +++ b/cli/util/fs.rs @@ -3,6 +3,7 @@ use deno_core::anyhow::Context; use deno_core::error::AnyError; pub use deno_core::normalize_path; +use deno_core::task::spawn_blocking; use deno_core::ModuleSpecifier; use deno_runtime::deno_crypto::rand; use deno_runtime::deno_node::PathClean; @@ -503,7 +504,7 @@ impl LaxSingleProcessFsFlag { // This uses a blocking task because we use a single threaded // runtime and this is time sensitive so we don't want it to update // at the whims of of whatever is occurring on the runtime thread. - tokio::task::spawn_blocking({ + spawn_blocking({ let token = token.clone(); let last_updated_path = last_updated_path.clone(); move || { |