diff --git a/packages/fullstack/Cargo.toml b/packages/fullstack/Cargo.toml index 57f414194..42f709872 100644 --- a/packages/fullstack/Cargo.toml +++ b/packages/fullstack/Cargo.toml @@ -41,6 +41,7 @@ log = { workspace = true } once_cell = "1.17.1" thiserror = { workspace = true } tokio = { workspace = true, features = ["full"], optional = true } +tokio-util = { version = "0.7.8", features = ["rt"], optional = true } object-pool = "0.5.4" anymap = "0.12.1" @@ -68,7 +69,7 @@ hot-reload = ["serde_json", "futures-util"] warp = ["dep:warp", "ssr"] axum = ["dep:axum", "tower-http", "ssr"] salvo = ["dep:salvo", "ssr"] -ssr = ["server_fn/ssr", "tokio", "dioxus-ssr", "tower", "hyper", "http", "http-body", "dioxus-router/ssr", "tokio-stream"] +ssr = ["server_fn/ssr", "tokio", "tokio-util", "dioxus-ssr", "tower", "hyper", "http", "http-body", "dioxus-router/ssr", "tokio-stream"] default-tls = ["server_fn/default-tls"] rustls = ["server_fn/rustls"] diff --git a/packages/fullstack/src/adapters/mod.rs b/packages/fullstack/src/adapters/mod.rs index d72a0323c..9133f8a08 100644 --- a/packages/fullstack/src/adapters/mod.rs +++ b/packages/fullstack/src/adapters/mod.rs @@ -17,12 +17,9 @@ pub mod salvo_adapter; #[cfg(feature = "warp")] pub mod warp_adapter; -use std::sync::{Arc, RwLock}; - use http::StatusCode; - use server_fn::{Encoding, Payload}; -use tokio::task::spawn_blocking; +use std::sync::{Arc, RwLock}; use crate::{ layer::{BoxedService, Service}, @@ -93,27 +90,22 @@ impl Service for ServerFnHandler { // Because the future returned by `server_fn_handler` is `Send`, and the future returned by this function must be send, we need to spawn a new runtime let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - spawn_blocking({ + let pool = get_local_pool(); + pool.spawn_pinned({ let function = function.clone(); let mut server_context = server_context.clone(); server_context.parts = parts; - move || { - tokio::runtime::Runtime::new() - .expect("couldn't spawn runtime") - .block_on(async move { - let data = match function.encoding() { - Encoding::Url | Encoding::Cbor => &body, - Encoding::GetJSON | Encoding::GetCBOR => &query, - }; - let server_function_future = function.call((), data); - let server_function_future = ProvideServerContext::new( - server_function_future, - server_context.clone(), - ); - let resp = server_function_future.await; + move || async move { + let data = match function.encoding() { + Encoding::Url | Encoding::Cbor => &body, + Encoding::GetJSON | Encoding::GetCBOR => &query, + }; + let server_function_future = function.call((), data); + let server_function_future = + ProvideServerContext::new(server_function_future, server_context.clone()); + let resp = server_function_future.await; - resp_tx.send(resp).unwrap(); - }) + resp_tx.send(resp).unwrap(); } }); let result = resp_rx.await.unwrap(); @@ -158,3 +150,17 @@ impl Service for ServerFnHandler { }) } } + +fn get_local_pool() -> tokio_util::task::LocalPoolHandle { + use once_cell::sync::OnceCell; + static LOCAL_POOL: OnceCell = OnceCell::new(); + LOCAL_POOL + .get_or_init(|| { + tokio_util::task::LocalPoolHandle::new( + std::thread::available_parallelism() + .map(Into::into) + .unwrap_or(1), + ) + }) + .clone() +}