async executor queue

Author: killzoner

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.2.0] - 2025-09-05
+### Changed
+- Use async executor to allow for long running CPU profile and allocs to run concurrently (works with pyroscope for example)
+
 ## [0.1.4] - 2025-08-20
 ### Fixed
 - Don't wait for the heap lock if already taken

Cargo.lock

@@ -3,7 +3,7 @@ authors = ["killzoner"]
 edition = "2024"
 name = "pprof_hyper_server"
 resolver = "3"
-version = "0.1.4"
+version = "0.2.0"
 
 description = "A minimal pprof server implementation using hyper without runtime dependency"
 documentation = "https://docs.rs/pprof_hyper_server"
@@ -14,11 +14,11 @@ repository = "https://github.com/killzoner/pprof-hyper-server"
 [dependencies]
 anyhow = { version = "1.0.99", default-features = false }
 async-channel = { version = "2.5.0", default-features = false }
+async-executor = { version = "1.13.3", default-features = false }
 async-io = { version = "2.5.0", default-features = false }
 form_urlencoded = { version = "1.2.2", default-features = false, features = [
   "alloc",
 ] }
-futures-lite = { version = "2.6.1", default-features = false }
 http-body-util = { version = "0.1.3", default-features = false }
 hyper = { version = "1.7.0", default-features = false, features = [
   "server",

Cargo.toml

@@ -7,8 +7,8 @@
 
 use anyhow::Result;
 use async_channel::bounded;
+use async_executor::Executor;
 use async_io::Async;
-use futures_lite::future;
 use http_body_util::Full;
 use hyper::{
     Method, Request, Response, StatusCode, Uri,
@@ -102,7 +102,6 @@ impl Task<'_> {
         );
 
         let blocklist = self.config.pprof_blocklist.unwrap_or(PPROF_BLOCKLIST);
-
         let guard = ProfilerGuardBuilder::default()
             .frequency(profile_sampling)
             .blocklist(blocklist)
@@ -183,14 +182,29 @@ pub async fn serve<'a>(bind_address: SocketAddr, config: Config<'a>) -> Result<(
     let listener = Async::<TcpListener>::bind(bind_address)?;
     let (s, r) = bounded::<Task>(MAX_CONCURRENT_REQUESTS);
     let config = Arc::new(config);
+    let ex = Arc::new(Executor::new());
 
-    loop {
-        // stack max MAX_CONCURRENT_REQUESTS requests, prefering stacking than answering to them.
-        // if we cannot stack anymore, drop the connection and other pending requests.
-        // we don't need a multi threaded server to serve pprof server, but don't want it to be a source of DDOS.
-        future::or(
-            async {
-                // Wait for a new client.
+    ex.spawn({
+        let ex = ex.clone();
+        async move {
+            loop {
+                if let Ok(task) = r.recv().await {
+                    ex.spawn(async {
+                        task.handle_client().await.unwrap_or_default();
+                    })
+                    .detach();
+                }
+            }
+        }
+    })
+    .detach();
+
+    ex.run({
+        async move {
+            // stack max MAX_CONCURRENT_REQUESTS requests
+            // if we cannot add more tasks, drop the connection
+            // we don't need a multi threaded server to serve pprof server, but don't want it to be a source of DOS.
+            loop {
                 let listener = listener.accept().await;
                 if let Ok((client, _)) = listener {
                     let task = Task {
@@ -201,13 +215,10 @@ pub async fn serve<'a>(bind_address: SocketAddr, config: Config<'a>) -> Result<(
                     // we ignore the potential error as it would mean we should drop the connection if channel is full.
                     let _ = s.try_send(task);
                 }
-            },
-            async {
-                if let Ok(task) = r.recv().await {
-                    task.handle_client().await.unwrap_or_default();
-                }
-            },
-        )
-        .await;
-    }
+            }
+        }
+    })
+    .await;
+
+    Ok(())
 }