feat: async phase handler

Add support for asynchronous phase handlers:
- Introduce async phase handler framework
- Add async request handling capabilities
- Implement async/await support for nginx phases
- Enable non-blocking request processing

Author: ensh63

Cargo.lock

@@ -48,6 +48,7 @@ targets = []
 [dependencies]
 allocator-api2 = { version = "0.4.0", default-features = false, features = ["fresh-rust"] }
 async-task = { version = "4.7.1", optional = true }
+futures-util = { version = "0.3", default-features = false }
 lock_api = "0.4.13"
 nginx-sys = { path = "nginx-sys", version = "0.5.0"}
 pin-project-lite = { version = "0.2.16", optional = true }

Cargo.toml

@@ -0,0 +1,162 @@
+use core::fmt::Display;
+use core::future::Future;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+use crate::http::{HttpModule, HttpPhase, HttpRequestHandler, IntoHandlerStatus, Request};
+use crate::{async_ as ngx_async, ngx_log_debug_http};
+
+use crate::ffi::{ngx_http_request_t, ngx_int_t, ngx_post_event, ngx_posted_events};
+
+use futures_util::FutureExt;
+use pin_project_lite::*;
+
+/// An asynchronous HTTP request handler trait.
+pub trait AsyncHandler {
+    /// The phase in which the handler will be executed.
+    const PHASE: HttpPhase;
+    /// The associated HTTP module type.
+    type Module: HttpModule;
+    /// The return type of the asynchronous worker function.
+    type Output: IntoHandlerStatus;
+    /// The asynchronous worker function to be implemented.
+    fn worker(request: &mut Request) -> impl Future<Output = Self::Output>;
+}
+
+const fn async_phase(phase: HttpPhase) -> HttpPhase {
+    assert!(
+        !matches!(phase, HttpPhase::Content),
+        "Content phase is not supported"
+    );
+    phase
+}
+
+/// An error type for asynchronous handler operations.
+#[derive(Debug)]
+pub enum AsyncHandlerError {
+    /// Indicates that the context creation failed.
+    ContextCreationFailed,
+    /// Indicates that there is no async launcher available.
+    NoAsyncLauncher,
+    /// Indicates that the context deletion failed.
+    ContextDeletionFailed,
+}
+
+impl Display for AsyncHandlerError {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            AsyncHandlerError::ContextCreationFailed => {
+                write!(f, "AsyncHandler: Context creation failed")
+            }
+            AsyncHandlerError::NoAsyncLauncher => {
+                write!(f, "AsyncHandler: No async launcher available")
+            }
+            AsyncHandlerError::ContextDeletionFailed => {
+                write!(f, "AsyncHandler: Context deletion failed")
+            }
+        }
+    }
+}
+
+#[derive(Default)]
+struct AsyncRequestContext {
+    launcher: Option<async_task::Task<ngx_int_t>>,
+}
+
+impl<AH> HttpRequestHandler for AH
+where
+    AH: AsyncHandler + 'static,
+{
+    const PHASE: HttpPhase = async_phase(AH::PHASE);
+    type Output = Result<ngx_int_t, AsyncHandlerError>;
+
+    fn handler(request: &mut Request) -> Self::Output {
+        let mut pool = request.pool();
+
+        let ctx = pool
+            .get_or_add_unique(|| {
+                let request_ptr: *mut ngx_http_request_t = request.as_mut() as *mut _ as _;
+                AsyncRequestContext {
+                    launcher: Some(ngx_async::spawn(handler_future::<AH>(request_ptr))),
+                }
+            })
+            .ok_or(AsyncHandlerError::ContextCreationFailed)?;
+
+        match &ctx.launcher {
+            None => Err(AsyncHandlerError::NoAsyncLauncher),
+            Some(launcher) if launcher.is_finished() => {
+                // task is finished, so both expect() should not panic
+                let task = ctx
+                    .launcher
+                    .take()
+                    .expect("AsyncHandler: Task should be present");
+                let rc = task
+                    .now_or_never()
+                    .expect("AsyncHandler: Task should be ready");
+                ngx_log_debug_http!(request, "AsyncHandler: task joined; rc = {}", rc);
+                pool.remove_unique::<AsyncRequestContext>()
+                    .ok_or(AsyncHandlerError::ContextDeletionFailed)?;
+                Ok(rc)
+            }
+            Some(_) => {
+                ngx_log_debug_http!(request, "AsyncHandler: running");
+                Ok(nginx_sys::NGX_AGAIN as _)
+            }
+        }
+    }
+}
+
+pin_project! {
+    struct HandlerFuture<Fut>
+    where
+        Fut: Future<Output = ngx_int_t>,
+    {
+        #[pin]
+        worker_fut: Fut,
+        request: *const ngx_http_request_t,
+    }
+}
+
+fn handler_future<AH>(request: *mut ngx_http_request_t) -> impl Future<Output = ngx_int_t>
+where
+    AH: AsyncHandler,
+{
+    let fut = async move {
+        let request = unsafe { Request::from_ngx_http_request(request) };
+        AH::worker(request).await.into_handler_status(request)
+    };
+
+    HandlerFuture::<_> {
+        worker_fut: fut,
+        request,
+    }
+}
+
+impl<Fut> Future for HandlerFuture<Fut>
+where
+    Fut: Future<Output = ngx_int_t>,
+{
+    type Output = ngx_int_t;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        let request = unsafe { Request::from_const_ngx_http_request(*this.request) };
+
+        match this.worker_fut.poll(cx) {
+            Poll::Pending => {
+                ngx_log_debug_http!(request, "HandlerFuture: pending");
+                Poll::Pending
+            }
+            Poll::Ready(rc) => {
+                unsafe {
+                    ngx_post_event(
+                        (*request.connection()).write,
+                        core::ptr::addr_of_mut!(ngx_posted_events),
+                    )
+                };
+                ngx_log_debug_http!(request, "HandlerFuture: ready");
+                Poll::Ready(rc)
+            }
+        }
+    }
+}

src/http/async_request.rs

@@ -1,10 +1,16 @@
+#[cfg(feature = "async")]
+mod async_request;
+
 mod conf;
 mod module;
 mod request;
 mod request_context;
 mod status;
 mod upstream;
 
+#[cfg(feature = "async")]
+pub use async_request::*;
+
 pub use conf::*;
 pub use module::*;
 pub use request::*;

src/http/mod.rs

@@ -9,6 +9,7 @@ use crate::core::*;
 use crate::ffi::*;
 use crate::http::HttpPhase;
 use crate::http::status::*;
+use crate::ngx_log_error;
 
 /// Define a static request handler.
 ///
@@ -85,7 +86,9 @@ macro_rules! http_variable_get {
 /// in the `into_handler_status` method.
 ///
 /// There are predefined implementations for `ngx_int_t`, [`Status`], [`HTTPStatus`],
-/// [`Option`] with value type implementing [`IntoHandlerStatus`].
+/// [`Option`] with value type implementing [`IntoHandlerStatus`],
+/// and [`Result`] with value type implementing [`IntoHandlerStatus`]
+/// and error type implementing [`core::fmt::Display`].
 pub trait IntoHandlerStatus
 where
     Self: Sized,
@@ -105,6 +108,23 @@ where
     }
 }
 
+impl<T, E> IntoHandlerStatus for Result<T, E>
+where
+    T: IntoHandlerStatus,
+    E: core::fmt::Display,
+{
+    #[inline]
+    fn into_handler_status(self, r: &Request) -> ngx_int_t {
+        match self {
+            Ok(val) => val.into_handler_status(r),
+            Err(e) => {
+                ngx_log_error!(NGX_LOG_ERR, r.log(), "{e}");
+                NGX_ERROR as _
+            }
+        }
+    }
+}
+
 impl IntoHandlerStatus for ngx_int_t {
     #[inline]
     fn into_handler_status(self, _r: &Request) -> ngx_int_t {
@@ -196,6 +216,16 @@ impl Request {
         unsafe { &mut *r.cast::<Request>() }
     }
 
+    /// Create a const [`Request`] from a const [`ngx_http_request_t`].
+    ///
+    /// # Safety
+    ///
+    /// The caller has provided a valid non-null pointer to a valid `ngx_http_request_t`
+    /// which shares the same representation as `Request`.
+    pub unsafe fn from_const_ngx_http_request<'a>(r: *const ngx_http_request_t) -> &'a Request {
+        unsafe { &*r.cast::<Request>() }
+    }
+
     /// Is this the main request (as opposed to a subrequest)?
     pub fn is_main(&self) -> bool {
         let main = self.0.main.cast();
@@ -301,6 +331,11 @@ impl Request {
         }
     }
 
+    /// Get HTTP status of response.
+    pub fn get_status(&self) -> HTTPStatus {
+        HTTPStatus::from_u16(self.0.headers_out.status as u16).unwrap_or(HTTPStatus(0))
+    }
+
     /// Set HTTP status of response.
     pub fn set_status(&mut self, status: HTTPStatus) {
         self.0.headers_out.status = status.into();
@@ -370,6 +405,14 @@ impl Request {
         unsafe { Status(ngx_http_output_filter(&raw mut self.0, body)) }
     }
 
+    /// Get the output chain buffer.
+    pub fn get_out(&self) -> Option<&ngx_chain_t> {
+        if self.0.out.is_null() {
+            return None;
+        }
+        unsafe { Some(&*self.0.out) }
+    }
+
     /// Perform internal redirect to a location
     pub fn internal_redirect(&self, location: &str) -> Status {
         assert!(!location.is_empty(), "uri location is empty");