diff --git a/.github/workflows/nginx.yaml b/.github/workflows/nginx.yaml index 2f73c744..aa02e9ca 100644 --- a/.github/workflows/nginx.yaml +++ b/.github/workflows/nginx.yaml @@ -49,10 +49,11 @@ env: NGX_TEST_FILES: examples/t NGX_TEST_GLOBALS_DYNAMIC: >- - load_module ${{ github.workspace }}/nginx/objs/ngx_http_async_module.so; + load_module ${{ github.workspace }}/nginx/objs/ngx_http_async_request_module.so; load_module ${{ github.workspace }}/nginx/objs/ngx_http_awssigv4_module.so; load_module ${{ github.workspace }}/nginx/objs/ngx_http_curl_module.so; load_module ${{ github.workspace }}/nginx/objs/ngx_http_shared_dict_module.so; + load_module ${{ github.workspace }}/nginx/objs/ngx_http_subrequest_module.so; load_module ${{ github.workspace }}/nginx/objs/ngx_http_upstream_custom_module.so; OPENSSL_VERSION: '3.0.16' diff --git a/Cargo.lock b/Cargo.lock index 7fed5944..0002d6ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "adler2" @@ -308,6 +308,30 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures-core" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" + +[[package]] +name = "futures-task" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" + +[[package]] +name = "futures-util" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "pin-utils", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -547,6 +571,7 @@ version = "0.5.0" dependencies = [ "allocator-api2", "async-task", + "futures-util", "lock_api", "nginx-sys", "pin-project-lite", @@ -623,6 +648,12 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "prettyplease" version = "0.2.37" diff --git a/Cargo.toml b/Cargo.toml index df0d190c..74994c5d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,6 +53,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 } diff --git a/examples/Cargo.toml b/examples/Cargo.toml index afde14b9..3ac3cc1b 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -13,6 +13,9 @@ autobins = false build = "../build.rs" [dependencies] +# allocator-api2 = { version = "0.4.0", default-features = false, features = ["fresh-rust"] } +# async-task = { version = "4.7.1" } +# futures = "0.3" nginx-sys = { path = "../nginx-sys/", default-features = false } ngx = { path = "../", default-features = false, features = ["std"] } @@ -47,15 +50,21 @@ path = "upstream.rs" crate-type = ["cdylib"] [[example]] -name = "async" -path = "async.rs" +name = "async_request" +path = "async_request.rs" crate-type = ["cdylib"] +required-features = ["async"] [[example]] name = "shared_dict" path = "shared_dict.rs" crate-type = ["cdylib"] +[[example]] +name = "subrequest" +path = "subrequest.rs" +crate-type = ["cdylib"] + [features] default = ["export-modules", "ngx/vendored"] # Generate `ngx_modules` table with module exports @@ -63,6 +72,7 @@ default = ["export-modules", "ngx/vendored"] # outside of the NGINX buildsystem. However, cargo currently does not detect # this configuration automatically. # See https://github.com/rust-lang/rust/issues/20267 +async = ["ngx/async"] export-modules = [] linux = [] diff --git a/examples/async.conf b/examples/async.conf deleted file mode 100644 index d96876e0..00000000 --- a/examples/async.conf +++ /dev/null @@ -1,24 +0,0 @@ -daemon off; -master_process off; -# worker_processes 1; - -load_module modules/libasync.so; -error_log error.log debug; - -events { } - -http { - server { - listen *:8000; - server_name localhost; - location / { - root html; - index index.html index.htm; - async on; - } - error_page 500 502 503 504 /50x.html; - location = /50x.html { - root html; - } - } -} diff --git a/examples/async.rs b/examples/async.rs deleted file mode 100644 index 15afa34a..00000000 --- a/examples/async.rs +++ /dev/null @@ -1,243 +0,0 @@ -extern crate alloc; - -use alloc::sync::Arc; -use core::ffi::{c_char, c_void}; -use core::mem; -use core::ptr; -use core::sync::atomic::{AtomicBool, AtomicPtr, Ordering}; -use core::time::Duration; -use std::sync::OnceLock; -use std::time::Instant; - -use ngx::core::Status; -use ngx::ffi::{ - NGX_CONF_TAKE1, NGX_HTTP_LOC_CONF, NGX_HTTP_LOC_CONF_OFFSET, NGX_HTTP_MODULE, NGX_LOG_EMERG, - ngx_command_t, ngx_conf_t, ngx_connection_t, ngx_event_t, ngx_http_module_t, ngx_int_t, - ngx_module_t, ngx_post_event, ngx_posted_events, ngx_posted_next_events, ngx_str_t, ngx_uint_t, -}; -use ngx::http::{self, HttpModule, HttpModuleLocationConf, HttpRequestHandler, MergeConfigError}; -use ngx::{ngx_conf_log_error, ngx_log_debug_http, ngx_string}; -use tokio::runtime::Runtime; - -struct Module; - -impl http::HttpModule for Module { - fn module() -> &'static ngx_module_t { - unsafe { &*::core::ptr::addr_of!(ngx_http_async_module) } - } - - unsafe extern "C" fn postconfiguration(cf: *mut ngx_conf_t) -> ngx_int_t { - // SAFETY: this function is called with non-NULL cf always - let cf = unsafe { &mut *cf }; - http::add_phase_handler::(cf) - .map_or(Status::NGX_ERROR, |_| Status::NGX_OK) - .into() - } -} - -#[derive(Debug, Default)] -struct ModuleConfig { - enable: bool, -} - -unsafe impl HttpModuleLocationConf for Module { - type LocationConf = ModuleConfig; -} - -static mut NGX_HTTP_ASYNC_COMMANDS: [ngx_command_t; 2] = [ - ngx_command_t { - name: ngx_string!("async"), - type_: (NGX_HTTP_LOC_CONF | NGX_CONF_TAKE1) as ngx_uint_t, - set: Some(ngx_http_async_commands_set_enable), - conf: NGX_HTTP_LOC_CONF_OFFSET, - offset: 0, - post: ptr::null_mut(), - }, - ngx_command_t::empty(), -]; - -static NGX_HTTP_ASYNC_MODULE_CTX: ngx_http_module_t = ngx_http_module_t { - preconfiguration: Some(Module::preconfiguration), - postconfiguration: Some(Module::postconfiguration), - create_main_conf: None, - init_main_conf: None, - create_srv_conf: None, - merge_srv_conf: None, - create_loc_conf: Some(Module::create_loc_conf), - merge_loc_conf: Some(Module::merge_loc_conf), -}; - -// Generate the `ngx_modules` table with exported modules. -// This feature is required to build a 'cdylib' dynamic module outside of the NGINX buildsystem. -#[cfg(feature = "export-modules")] -ngx::ngx_modules!(ngx_http_async_module); - -#[used] -#[allow(non_upper_case_globals)] -#[cfg_attr(not(feature = "export-modules"), unsafe(no_mangle))] -pub static mut ngx_http_async_module: ngx_module_t = ngx_module_t { - ctx: &raw const NGX_HTTP_ASYNC_MODULE_CTX as _, - commands: unsafe { &raw mut NGX_HTTP_ASYNC_COMMANDS[0] }, - type_: NGX_HTTP_MODULE as _, - ..ngx_module_t::default() -}; - -impl http::Merge for ModuleConfig { - fn merge(&mut self, prev: &ModuleConfig) -> Result<(), MergeConfigError> { - if prev.enable { - self.enable = true; - }; - Ok(()) - } -} - -unsafe extern "C" fn check_async_work_done(event: *mut ngx_event_t) { - let ctx = ngx::ngx_container_of!(event, RequestCTX, event); - let c: *mut ngx_connection_t = unsafe { (*event).data.cast() }; - - if unsafe { (*ctx).done.load(Ordering::Relaxed) } { - // Triggering async_access_handler again - unsafe { ngx_post_event((*c).write, &raw mut ngx_posted_events) }; - } else { - // this doesn't have have good performance but works as a simple thread-safe example and - // doesn't causes segfault. The best method that provides both thread-safety and - // performance requires an nginx patch. - unsafe { ngx_post_event(event, &raw mut ngx_posted_next_events) }; - } -} - -struct RequestCTX { - done: Arc, - event: ngx_event_t, - task: Option>, -} - -impl Default for RequestCTX { - fn default() -> Self { - Self { - done: AtomicBool::new(false).into(), - event: unsafe { mem::zeroed() }, - task: Default::default(), - } - } -} - -impl Drop for RequestCTX { - fn drop(&mut self) { - if let Some(handle) = self.task.take() { - handle.abort(); - } - - if self.event.posted() != 0 { - unsafe { ngx::ffi::ngx_delete_posted_event(&raw mut self.event) }; - } - } -} - -struct AsyncAccessHandler; - -impl HttpRequestHandler for AsyncAccessHandler { - const PHASE: ngx::http::HttpPhase = ngx::http::HttpPhase::Access; - type Output = Status; - - fn handler(request: &mut http::Request) -> Self::Output { - let co = Module::location_conf(request).expect("module config is none"); - - ngx_log_debug_http!(request, "async module enabled: {}", co.enable); - - if !co.enable { - return Status::NGX_DECLINED; - } - - if let Some(ctx) = request.get_module_ctx::(Module::module()) { - if !ctx.done.load(Ordering::Relaxed) { - return Status::NGX_AGAIN; - } - - return Status::NGX_OK; - } - - let ctx = request.pool().allocate(RequestCTX::default()); - if ctx.is_null() { - return Status::NGX_ERROR; - } - request.set_module_ctx(ctx.cast(), Module::module()); - - let ctx = unsafe { &mut *ctx }; - ctx.event.handler = Some(check_async_work_done); - ctx.event.data = request.connection().cast(); - ctx.event.log = unsafe { (*request.connection()).log }; - unsafe { ngx_post_event(&raw mut ctx.event, &raw mut ngx_posted_next_events) }; - - // Request is no longer needed and can be converted to something movable to the async block - let req = AtomicPtr::new(request.into()); - let done_flag = ctx.done.clone(); - - let rt = ngx_http_async_runtime(); - ctx.task = Some(rt.spawn(async move { - let start = Instant::now(); - tokio::time::sleep(Duration::from_secs(2)).await; - let req = unsafe { http::Request::from_ngx_http_request(req.load(Ordering::Relaxed)) }; - // not really thread safe, we should apply all these operation in nginx thread - // but this is just an example. proper way would be storing these headers in the request - // ctx and apply them when we get back to the nginx thread. - req.add_header_out( - "X-Async-Time", - start.elapsed().as_millis().to_string().as_str(), - ); - - done_flag.store(true, Ordering::Release); - // there is a small issue here. If traffic is low we may get stuck behind a 300ms timer - // in the nginx event loop. To workaround it we can notify the event loop using - // pthread_kill( nginx_thread, SIGIO ) to wake up the event loop. (or patch nginx - // and use the same trick as the thread pool) - })); - - Status::NGX_AGAIN - } -} - -extern "C" fn ngx_http_async_commands_set_enable( - cf: *mut ngx_conf_t, - _cmd: *mut ngx_command_t, - conf: *mut c_void, -) -> *mut c_char { - unsafe { - let conf = &mut *(conf as *mut ModuleConfig); - let args: &[ngx_str_t] = (*(*cf).args).as_slice(); - let val = match args[1].to_str() { - Ok(s) => s, - Err(_) => { - ngx_conf_log_error!(NGX_LOG_EMERG, cf, "`async` argument is not utf-8 encoded"); - return ngx::core::NGX_CONF_ERROR; - } - }; - - // set default value optionally - conf.enable = false; - - if val.eq_ignore_ascii_case("on") { - conf.enable = true; - } else if val.eq_ignore_ascii_case("off") { - conf.enable = false; - } - }; - - ngx::core::NGX_CONF_OK -} - -fn ngx_http_async_runtime() -> &'static Runtime { - // Should not be called from the master process - assert_ne!( - unsafe { ngx::ffi::ngx_process }, - ngx::ffi::NGX_PROCESS_MASTER as _ - ); - - static RUNTIME: OnceLock = OnceLock::new(); - RUNTIME.get_or_init(|| { - tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build() - .expect("tokio runtime init") - }) -} diff --git a/examples/async_request.rs b/examples/async_request.rs new file mode 100644 index 00000000..710fbded --- /dev/null +++ b/examples/async_request.rs @@ -0,0 +1,243 @@ +use ngx::http::{ + AsyncHandler, HTTPStatus, HttpModule, HttpModuleLocationConf, HttpPhase, Merge, + MergeConfigError, Request, add_phase_handler, +}; + +use ngx::http::subrequest::{SR_MODIFIER_NO_OP, SubRequestBuilder, SubRequestError}; +use ngx::{async_ as ngx_async, ngx_log_debug_http, ngx_log_error}; + +use nginx_sys::{ + NGX_CONF_TAKE1, NGX_HTTP_LOC_CONF, NGX_HTTP_LOC_CONF_OFFSET, ngx_command_t, ngx_conf_t, + ngx_flag_t, ngx_http_complex_value_t, ngx_http_module_t, ngx_http_request_t, + ngx_http_send_response, ngx_int_t, ngx_module_t, ngx_str_t, ngx_uint_t, +}; + +const NGX_CONF_UNSET_FLAG: ngx_flag_t = nginx_sys::NGX_CONF_UNSET as _; + +struct SampleAsyncHandler; + +enum SampleAsyncHandlerError { + SubRequestCreationFailed(SubRequestError), + SubRequestFailed(ngx_int_t), + NoSubRequestReturned, +} + +impl core::fmt::Display for SampleAsyncHandlerError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + SampleAsyncHandlerError::SubRequestCreationFailed(e) => { + write!(f, "subrequest creation failed: {}", e) + } + SampleAsyncHandlerError::SubRequestFailed(rc) => { + write!(f, "subrequest failed with return code: {}", rc) + } + SampleAsyncHandlerError::NoSubRequestReturned => { + write!(f, "subrequest did not return a request reference") + } + } + } +} + +impl From for SampleAsyncHandlerError { + fn from(err: SubRequestError) -> Self { + SampleAsyncHandlerError::SubRequestCreationFailed(err) + } +} + +impl From for SampleAsyncHandlerError { + fn from(rc: ngx_int_t) -> Self { + SampleAsyncHandlerError::SubRequestFailed(rc) + } +} + +impl AsyncHandler for SampleAsyncHandler { + const PHASE: HttpPhase = HttpPhase::Access; + type Module = Module; + type Output = Result; + + async fn worker(request: &mut Request) -> Self::Output { + ngx_log_debug_http!(request, "worker started"); + + let co = Module::location_conf(request).expect("module config is none"); + ngx_log_debug_http!(request, "async_request module enabled: {}", co.enable); + + if co.enable != 1 { + return Ok(nginx_sys::NGX_DECLINED as _); + } + + let log = request.log(); + let request_ptr: *mut ngx_http_request_t = request.as_mut(); + let uri: &str = if co.uri.is_empty() { + "/proxy" + } else { + co.uri.to_str().unwrap_or("/proxy") + }; + + let mut sr: Option<&Request> = None; + + let subrc = SubRequestBuilder::new(request.pool(), uri)? + .args("arg1=val1&arg2=val2")? + .in_memory() + .waited() + .build_async(request, SR_MODIFIER_NO_OP, |r, rc| { + sr = Some(r); + rc + }) + .await?; + + ngx_log_error!(nginx_sys::NGX_LOG_INFO, log, "subrequest rc:{}", subrc); + + if subrc != nginx_sys::NGX_OK as _ { + return HTTPStatus::try_from(subrc) + .map(Into::into) + .map_err(|_| SampleAsyncHandlerError::from(subrc)); + } + + if sr.is_none() { + return Err(SampleAsyncHandlerError::NoSubRequestReturned); + } + + let sr = sr.unwrap(); + + ngx_log_error!( + nginx_sys::NGX_LOG_INFO, + log, + "Subrequest status: {:?}", + sr.get_status() + ); + + ngx_async::sleep(core::time::Duration::from_millis(100)).await; + + let mut resp_len: usize = 0; + + let mut rc = nginx_sys::NGX_OK as ngx_int_t; + + if let Some(out) = sr.get_out() { + if !out.buf.is_null() { + let b = unsafe { &*out.buf }; + resp_len = unsafe { b.last.offset_from(b.pos) } as usize; + + let sr_ptr: *const ngx_http_request_t = sr.as_ref(); + + let mut ct: ngx_str_t = (unsafe { *sr_ptr }).headers_out.content_type; + + let mut cv: ngx_http_complex_value_t = unsafe { core::mem::zeroed() }; + cv.value = ngx_str_t { + len: resp_len as _, + data: b.pos as _, + }; + + rc = unsafe { + ngx_http_send_response(request_ptr, sr.get_status().0, &raw mut ct, &raw mut cv) + }; + + if rc == nginx_sys::NGX_OK as _ { + rc = nginx_sys::NGX_HTTP_OK as _; + } + } + } + + ngx_log_error!( + nginx_sys::NGX_LOG_INFO, + log, + "async handler after timeout; subrequest response length: {}", + resp_len + ); + + Ok(rc) + } +} + +static NGX_HTTP_ASYNC_REQUEST_MODULE_CTX: ngx_http_module_t = ngx_http_module_t { + preconfiguration: None, + postconfiguration: Some(Module::postconfiguration), + create_main_conf: None, + init_main_conf: None, + create_srv_conf: None, + merge_srv_conf: None, + create_loc_conf: Some(Module::create_loc_conf), + merge_loc_conf: Some(Module::merge_loc_conf), +}; + +#[cfg(feature = "export-modules")] +ngx::ngx_modules!(ngx_http_async_request_module); + +#[used] +#[allow(non_upper_case_globals)] +#[cfg_attr(not(feature = "export-modules"), unsafe(no_mangle))] +pub static mut ngx_http_async_request_module: ngx_module_t = ngx_module_t { + ctx: &raw const NGX_HTTP_ASYNC_REQUEST_MODULE_CTX as _, + commands: unsafe { &raw mut NGX_HTTP_ASYNC_REQUEST_COMMANDS[0] }, + type_: nginx_sys::NGX_HTTP_MODULE as _, + ..ngx_module_t::default() +}; + +struct Module; + +impl HttpModule for Module { + fn module() -> &'static ngx_module_t { + unsafe { &*::core::ptr::addr_of!(ngx_http_async_request_module) } + } + + unsafe extern "C" fn postconfiguration(cf: *mut ngx_conf_t) -> ngx_int_t { + // SAFETY: this function is called with non-NULL cf always + let cf = unsafe { &mut *cf }; + add_phase_handler::(cf) + .map_or(nginx_sys::NGX_ERROR as _, |_| nginx_sys::NGX_OK as _) + } +} + +#[derive(Debug)] +struct ModuleConfig { + enable: ngx_flag_t, + uri: ngx_str_t, +} + +impl Default for ModuleConfig { + fn default() -> Self { + Self { + enable: NGX_CONF_UNSET_FLAG, + uri: ngx_str_t::empty(), + } + } +} + +impl Merge for ModuleConfig { + fn merge(&mut self, prev: &ModuleConfig) -> Result<(), MergeConfigError> { + if self.enable == NGX_CONF_UNSET_FLAG { + if prev.enable != NGX_CONF_UNSET_FLAG { + self.enable = prev.enable; + } else { + self.enable = 0; + } + } + if self.uri.len == 0 { + self.uri = prev.uri; + } + Ok(()) + } +} + +unsafe impl HttpModuleLocationConf for Module { + type LocationConf = ModuleConfig; +} + +static mut NGX_HTTP_ASYNC_REQUEST_COMMANDS: [ngx_command_t; 3] = [ + ngx_command_t { + name: ngx::ngx_string!("async_request"), + type_: (NGX_HTTP_LOC_CONF | NGX_CONF_TAKE1) as ngx_uint_t, + set: Some(nginx_sys::ngx_conf_set_flag_slot), + conf: NGX_HTTP_LOC_CONF_OFFSET, + offset: core::mem::offset_of!(ModuleConfig, enable), + post: core::ptr::null_mut(), + }, + ngx_command_t { + name: ngx::ngx_string!("async_uri"), + type_: (NGX_HTTP_LOC_CONF | NGX_CONF_TAKE1) as ngx_uint_t, + set: Some(nginx_sys::ngx_conf_set_str_slot), + conf: NGX_HTTP_LOC_CONF_OFFSET, + offset: core::mem::offset_of!(ModuleConfig, uri), + post: core::ptr::null_mut(), + }, + ngx_command_t::empty(), +]; diff --git a/examples/config b/examples/config index 6b763652..182c2138 100644 --- a/examples/config +++ b/examples/config @@ -16,11 +16,14 @@ if [ $HTTP = YES ]; then ngx_rust_target_features= if :; then - ngx_module_name=ngx_http_async_module - ngx_module_libs="-lm" - ngx_rust_target_name=async + ngx_module_name=ngx_http_async_request_module + ngx_module_libs= + ngx_rust_target_name=async_request + ngx_rust_target_features=async ngx_rust_module + + ngx_rust_target_features= fi if :; then @@ -47,6 +50,14 @@ if [ $HTTP = YES ]; then ngx_rust_module fi + if :; then + ngx_module_name=ngx_http_subrequest_module + ngx_module_libs= + ngx_rust_target_name=subrequest + + ngx_rust_module + fi + if :; then ngx_module_name=ngx_http_upstream_custom_module ngx_module_libs= diff --git a/examples/subrequest.rs b/examples/subrequest.rs new file mode 100644 index 00000000..5bed1511 --- /dev/null +++ b/examples/subrequest.rs @@ -0,0 +1,257 @@ +use core::fmt::Display; + +use ngx::core::Status; +use ngx::http::subrequest::{SubRequestBuilder, SubRequestError}; +use ngx::http::{ + HTTPStatus, HttpModule, HttpModuleLocationConf, HttpPhase, HttpRequestHandler, Merge, + MergeConfigError, Request, RequestContext, add_phase_handler, +}; +use ngx::{ngx_log_debug_http, ngx_log_error}; + +use nginx_sys::{ + NGX_CONF_TAKE1, NGX_ERROR, NGX_HTTP_LOC_CONF, NGX_HTTP_LOC_CONF_OFFSET, ngx_command_t, + ngx_conf_t, ngx_flag_t, ngx_http_complex_value_t, ngx_http_module_t, ngx_http_request_t, + ngx_http_send_response, ngx_int_t, ngx_module_t, ngx_str_t, ngx_uint_t, +}; + +const NGX_CONF_UNSET_FLAG: ngx_flag_t = nginx_sys::NGX_CONF_UNSET as _; + +struct SampleHandler; + +enum SampleHandlerError { + ContextAllocation, + SubRequestCreation(SubRequestError), + SubRequest(ngx_int_t), + Response(ngx_int_t), +} + +impl From for SampleHandlerError { + fn from(e: SubRequestError) -> Self { + SampleHandlerError::SubRequestCreation(e) + } +} + +impl Display for SampleHandlerError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + SampleHandlerError::ContextAllocation => { + write!(f, "context allocation failed") + } + SampleHandlerError::SubRequestCreation(e) => { + write!(f, "subrequest creation failed: {}", e) + } + SampleHandlerError::SubRequest(rc) => { + write!(f, "subrequest failed with return code: {}", rc) + } + SampleHandlerError::Response(rc) => { + write!(f, "response creation failed with return code: {}", rc) + } + } + } +} + +impl HttpRequestHandler for SampleHandler { + const PHASE: HttpPhase = HttpPhase::Access; + type Output = Result; + + fn handler(request: &mut Request) -> Self::Output { + let co = Module::location_conf(request).expect("module config is none"); + ngx_log_debug_http!(request, "subrequest module enabled: {}", co.enable); + + if co.enable != 1 { + return Ok(Status::NGX_DECLINED); + } + + let rptr: *mut ngx_http_request_t = request.as_mut(); + + match SRCtx::get(request) { + Some(ctx) => ctx.rc.map_or( + // `ctx` has been created but not filled yet - subrequest is still in progress + Ok(Status::NGX_AGAIN), + // `ctx` has been created and filled - subrequest is completed + |rc| { + let status = ctx.status.0; + let msg = format!("subrequest completed with HTTP status: {status}, rc: {rc}"); + ngx_log_debug_http!(request, "{msg}"); + + if ctx.status.0 >= nginx_sys::NGX_HTTP_SPECIAL_RESPONSE as _ { + Ok(Status::from(ctx.status)) + } else if rc == nginx_sys::NGX_OK as _ && ctx.out.is_some() { + let outbuf = unsafe { &*ctx.out.unwrap().buf }; + let mut ct = ctx.ct; + let mut cv: ngx_http_complex_value_t = unsafe { core::mem::zeroed() }; + cv.value = ngx_str_t { + len: unsafe { outbuf.last.offset_from(outbuf.pos) } as _, + data: outbuf.pos as _, + }; + let resp_rc = unsafe { + ngx_http_send_response(rptr, status, &raw mut ct, &raw mut cv) + }; + if resp_rc == nginx_sys::NGX_OK as _ { + Ok(Status::from(ctx.status)) + } else { + Err(SampleHandlerError::Response(resp_rc)) + } + } else if let Ok(http_status) = HTTPStatus::try_from(rc) { + Ok(Status::from(http_status)) + } else { + Err(SampleHandlerError::SubRequest(rc)) + } + }, + ), + None => { + if SRCtx::create(request, SRCtx::default).is_some() { + let uri: &str = if co.uri.is_empty() { + "/proxy" + } else { + co.uri.to_str().unwrap_or("/proxy") + }; + + SubRequestBuilder::new(request.pool(), uri)? + .args("arg1=val1&arg2=val2")? + .in_memory() + .waited() + .build(request, sr_handler)?; + + Ok(Status::NGX_AGAIN) + } else { + Err(SampleHandlerError::ContextAllocation) + } + } + } + } +} + +struct SRCtx<'r> { + rc: Option, + status: HTTPStatus, + out: Option<&'r nginx_sys::ngx_chain_t>, + ct: ngx_str_t, +} + +impl Default for SRCtx<'_> { + fn default() -> Self { + Self { + rc: None, + status: HTTPStatus(NGX_ERROR as _), + out: None, + ct: ngx_str_t::empty(), + } + } +} + +impl RequestContext for SRCtx<'_> {} + +fn sr_handler(r: &mut Request, mut rc: ngx_int_t) -> ngx_int_t { + let newctx = SRCtx { + rc: Some(rc), + status: r.get_status(), + out: core::ptr::NonNull::new(r.as_ref().out).map(|out| unsafe { out.as_ref() }), + ct: r.as_ref().headers_out.content_type, + }; + if let Some(ctx) = SRCtx::get_mut(r.get_main_mut()) { + *ctx = newctx; + } else { + ngx_log_error!( + nginx_sys::NGX_LOG_ERR, + r.log(), + "subrequest: context not found" + ); + rc = NGX_ERROR as _; + } + rc +} + +static NGX_HTTP_SUBREQUEST_MODULE_CTX: ngx_http_module_t = ngx_http_module_t { + preconfiguration: None, + postconfiguration: Some(Module::postconfiguration), + create_main_conf: None, + init_main_conf: None, + create_srv_conf: None, + merge_srv_conf: None, + create_loc_conf: Some(Module::create_loc_conf), + merge_loc_conf: Some(Module::merge_loc_conf), +}; + +#[cfg(feature = "export-modules")] +ngx::ngx_modules!(ngx_http_subrequest_module); + +#[used] +#[allow(non_upper_case_globals)] +#[cfg_attr(not(feature = "export-modules"), unsafe(no_mangle))] +pub static mut ngx_http_subrequest_module: ngx_module_t = ngx_module_t { + ctx: &raw const NGX_HTTP_SUBREQUEST_MODULE_CTX as _, + commands: unsafe { &raw mut NGX_HTTP_SUBREQUEST_COMMANDS[0] }, + type_: nginx_sys::NGX_HTTP_MODULE as _, + ..ngx_module_t::default() +}; + +struct Module; + +impl HttpModule for Module { + fn module() -> &'static ngx_module_t { + unsafe { &*::core::ptr::addr_of!(ngx_http_subrequest_module) } + } + + unsafe extern "C" fn postconfiguration(cf: *mut ngx_conf_t) -> ngx_int_t { + // SAFETY: this function is called with non-NULL cf always + let cf = unsafe { &mut *cf }; + add_phase_handler::(cf) + .map_or(nginx_sys::NGX_ERROR as _, |_| nginx_sys::NGX_OK as _) + } +} + +#[derive(Debug)] +struct ModuleConfig { + enable: ngx_flag_t, + uri: ngx_str_t, +} + +impl Default for ModuleConfig { + fn default() -> Self { + Self { + enable: NGX_CONF_UNSET_FLAG, + uri: ngx_str_t::empty(), + } + } +} + +impl Merge for ModuleConfig { + fn merge(&mut self, prev: &ModuleConfig) -> Result<(), MergeConfigError> { + if self.enable == NGX_CONF_UNSET_FLAG { + if prev.enable != NGX_CONF_UNSET_FLAG { + self.enable = prev.enable; + } else { + self.enable = 0; + } + } + if self.uri.len == 0 { + self.uri = prev.uri; + } + Ok(()) + } +} + +unsafe impl HttpModuleLocationConf for Module { + type LocationConf = ModuleConfig; +} + +static mut NGX_HTTP_SUBREQUEST_COMMANDS: [ngx_command_t; 3] = [ + ngx_command_t { + name: ngx::ngx_string!("subrequest"), + type_: (NGX_HTTP_LOC_CONF | NGX_CONF_TAKE1) as ngx_uint_t, + set: Some(nginx_sys::ngx_conf_set_flag_slot), + conf: NGX_HTTP_LOC_CONF_OFFSET, + offset: core::mem::offset_of!(ModuleConfig, enable), + post: core::ptr::null_mut(), + }, + ngx_command_t { + name: ngx::ngx_string!("subrequest_uri"), + type_: (NGX_HTTP_LOC_CONF | NGX_CONF_TAKE1) as ngx_uint_t, + set: Some(nginx_sys::ngx_conf_set_str_slot), + conf: NGX_HTTP_LOC_CONF_OFFSET, + offset: core::mem::offset_of!(ModuleConfig, uri), + post: core::ptr::null_mut(), + }, + ngx_command_t::empty(), +]; diff --git a/examples/t/async.t b/examples/t/async.t deleted file mode 100644 index 98505fdb..00000000 --- a/examples/t/async.t +++ /dev/null @@ -1,56 +0,0 @@ -#!/usr/bin/perl - -# (C) Nginx, Inc - -# Tests for ngx-rust example modules. - -############################################################################### - -use warnings; -use strict; - -use Test::More; - -BEGIN { use FindBin; chdir($FindBin::Bin); } - -use lib 'lib'; -use Test::Nginx; - -############################################################################### - -select STDERR; $| = 1; -select STDOUT; $| = 1; - -my $t = Test::Nginx->new()->has(qw/http/)->plan(1) - ->write_file_expand('nginx.conf', <<'EOF'); - -%%TEST_GLOBALS%% - -daemon off; - -events { -} - -http { - %%TEST_GLOBALS_HTTP%% - - server { - listen 127.0.0.1:8080; - server_name localhost; - - location / { - async on; - } - } -} - -EOF - -$t->write_file('index.html', ''); -$t->run(); - -############################################################################### - -like(http_get('/index.html'), qr/X-Async-Time:/, 'async handler'); - -############################################################################### diff --git a/examples/t/async_request.t b/examples/t/async_request.t new file mode 100644 index 00000000..2cd6b6a7 --- /dev/null +++ b/examples/t/async_request.t @@ -0,0 +1,79 @@ +#!/usr/bin/perl + +# (C) Nginx, Inc + +# Tests for ngx-rust example modules. + +############################################################################### + +use warnings; +use strict; + +use Test::More; + +BEGIN { use FindBin; chdir($FindBin::Bin); } + +use lib 'lib'; +use Test::Nginx; + +############################################################################### + +select STDERR; $| = 1; +select STDOUT; $| = 1; + +my $t = Test::Nginx->new()->has(qw/http proxy/)->plan(2) + ->write_file_expand('nginx.conf', <<"EOF"); + +%%TEST_GLOBALS%% + +daemon off; + +events { +} + +http { + %%TEST_GLOBALS_HTTP%% + + server { + listen 127.0.0.1:8080; + server_name localhost; + + location / { + async_request on; + async_uri /proxy; + } + + location /non_existing { + async_request on; + async_uri /non_existing_upstream; + } + + location /timeout { + async_request on; + async_uri /slow_proxy; + } + + location /proxy { + internal; + proxy_pass http://127.0.0.1:8081; + } + } + + server { + listen 127.0.0.1:8081; + server_name localhost; + + location / { + return 200 'Hello from backend'; + } + } +} + +EOF + +$t->write_file('index.html', ''); +$t->run(); + +like(http_get('/'), qr/200 OK.*Hello from backend/s, 'async subrequest'); +like(http_get('/non_existing'), qr/404 Not Found/s, + 'async subrequest to non-existing upstream'); diff --git a/examples/t/subrequest.t b/examples/t/subrequest.t new file mode 100644 index 00000000..545c3773 --- /dev/null +++ b/examples/t/subrequest.t @@ -0,0 +1,76 @@ +#!/usr/bin/perl + +# (C) Nginx, Inc + +# Tests for ngx-rust example modules. + +############################################################################### + +use warnings; +use strict; + +use Test::More; + +BEGIN { use FindBin; chdir($FindBin::Bin); } + +use lib 'lib'; +use Test::Nginx; + +############################################################################### + +select STDERR; $| = 1; +select STDOUT; $| = 1; + +my $t = Test::Nginx->new()->has(qw/http proxy/)->plan(2) + ->write_file_expand('nginx.conf', <<"EOF"); + +%%TEST_GLOBALS%% + +daemon off; + +events { +} + +http { + %%TEST_GLOBALS_HTTP%% + + server { + listen 127.0.0.1:8080; + server_name localhost; + + location / { + subrequest on; + subrequest_uri /proxy; + } + + location /non_existing { + subrequest on; + subrequest_uri /non_existing_upstream; + } + + location /proxy { + internal; + proxy_pass http://127.0.0.1:8081; + } + } + + server { + listen 127.0.0.1:8081; + server_name localhost; + + location / { + return 200 'Hello from backend'; + } + } +} + +EOF + +$t->write_file('index.html', ''); +$t->run(); + +like(http_get('/'), + qr/200 OK.*Hello from backend/s, + 'subrequest'); +like(http_get('/non_existing'), qr/404 Not Found/s, + 'subrequest to non-existing upstream'); diff --git a/src/core/pool.rs b/src/core/pool.rs index 281bbbf5..c8829882 100644 --- a/src/core/pool.rs +++ b/src/core/pool.rs @@ -5,7 +5,7 @@ use core::ptr::{self, NonNull}; use nginx_sys::{ NGX_ALIGNMENT, ngx_buf_t, ngx_create_temp_buf, ngx_palloc, ngx_pcalloc, ngx_pfree, - ngx_pmemalign, ngx_pnalloc, ngx_pool_cleanup_add, ngx_pool_t, + ngx_pmemalign, ngx_pnalloc, ngx_pool_cleanup_add, ngx_pool_cleanup_t, ngx_pool_t, }; use crate::allocator::{AllocError, Allocator, dangling_for_layout}; @@ -134,6 +134,11 @@ impl AsMut for Pool { } } +// Wrapper to create an unique value type +struct Item { + value: T, +} + impl Pool { /// Creates a new `Pool` from an `ngx_pool_t` pointer. /// @@ -206,25 +211,98 @@ impl Pool { Some(MemoryBuffer::from_ngx_buf(buf)) } - /// Adds a cleanup handler for a value in the memory pool. + /// Allocates memory for a value and adds a cleanup handler to the memory pool. /// - /// Returns `Ok(())` if the cleanup handler is successfully added, or `Err(())` if the cleanup - /// handler cannot be added. + /// The value is created by calling the provided closure `f`. If allocation fails, + /// the closure is not called. + /// + /// Returns `Some(NonNull)` if the allocation and cleanup handler addition are successful, + /// or `None` if allocation fails. /// /// # Safety /// This function is marked as unsafe because it involves raw pointer manipulation. - unsafe fn add_cleanup_for_value(&self, value: *mut T) -> Result<(), ()> { - let cln = unsafe { ngx_pool_cleanup_add(self.0.as_ptr(), 0) }; + /// The returned pointer must not outlive the pool, must not be freed manually + /// (as it has a cleanup handler), and must not be accessed after the pool is destroyed. + pub unsafe fn allocate_with_cleanup T>( + &self, + f: F, + ) -> Option> { + let cln = unsafe { ngx_pool_cleanup_add(self.0.as_ptr(), mem::size_of::()) }; if cln.is_null() { - return Err(()); + return None; } - unsafe { + // 'data' may be NULL only if `T` is zero-sized. In that case, no real value is stored, + // so we can just use the cleanup structure itself as a placeholder. + // Note that zero-sized `T` may implement `Drop`, and this implementation will be + // called at cleanup time. + if (*cln).data.is_null() { + (*cln).data = cln as _; + }; (*cln).handler = Some(cleanup_type::); - (*cln).data = value as *mut c_void; + // `data` points to the memory allocated for the value by `ngx_pool_cleanup_add()` + ptr::write((*cln).data as *mut T, f()); + + NonNull::new((*cln).data as *mut T) + } + } + + /// Runs the cleanup handler for a value and removes it from the cleanup chain. + /// + /// If `value` is `Some`, removes the specific value's cleanup handler. + /// If `value` is `None`, removes the first cleanup handler found for type `T`. + /// + /// Returns `Some(())` if a cleanup handler was found and executed, + /// or `None` if no matching cleanup handler was found. + /// + /// # Safety + /// The caller must ensure that if `value` is `Some`, it points to a valid value + /// that has an associated cleanup handler in the pool. + unsafe fn remove_cleanup(&self, value: Option<*const T>) -> Option<()> { + unsafe { + self.cleanup_lookup::(value).map(|mut cln| { + let cln = cln.as_mut(); + cln.handler.take().inspect(|handler| { + handler(cln.data); + }); + cln.data = core::ptr::null_mut(); + }) } + } - Ok(()) + /// Searches for a cleanup handler in the pool's cleanup chain. + /// + /// If `value` is `Some`, searches for the cleanup handler associated with that specific value. + /// If `value` is `None`, returns the first cleanup handler found for type `T`. + /// + /// Returns `Some(NonNull)` if a matching cleanup handler is found, + /// or `None` if no matching handler is found. + /// + /// # Safety + /// This function is marked as unsafe because it involves raw pointer manipulation + /// and traverses the nginx cleanup chain structure. + unsafe fn cleanup_lookup( + &self, + value: Option<*const T>, + ) -> Option> { + let mut cln = (unsafe { *self.0.as_ptr() }).cleanup; + + while !cln.is_null() { + // SAFETY: comparing function pointers is generally unreliable, but in this specific + // case we can assume that the same function pointer was used when adding the cleanup + // handler. + unsafe { + #[allow(unpredictable_function_pointer_comparisons)] + if (*cln).handler == Some(cleanup_type::) + && (value.is_none() || (*cln).data == value.unwrap() as *mut c_void) + { + return NonNull::new(cln); + } + cln = (*cln).next; + } + } + + None } /// Allocates memory from the pool of the specified size. @@ -278,18 +356,87 @@ impl Pool { /// /// Returns a typed pointer to the allocated memory if successful, or a null pointer if /// allocation or cleanup handler addition fails. - pub fn allocate(&self, value: T) -> *mut T { + pub fn allocate(&self, value: T) -> *mut T { unsafe { - let p = self.alloc(mem::size_of::()) as *mut T; - ptr::write(p, value); - if self.add_cleanup_for_value(p).is_err() { - ptr::drop_in_place(p); - return ptr::null_mut(); - }; - p + match self.allocate_with_cleanup(|| value) { + None => ptr::null_mut(), + Some(mut ptr) => ptr.as_mut(), + } } } + /// Gets the unique value of type `T` from the memory pool, or allocates and initializes it + /// using the provided function if it does not exist. + /// + /// This ensures only one value of type `T` exists in the pool. If a value already exists, + /// it is returned and the function `f` is not called. If no value exists, a new one is + /// created using `f` and stored with a cleanup handler. If allocation fails, `f` is not called. + /// + /// Returns a mutable reference to the value if successful, or `None` if allocation fails. + pub fn get_or_add_unique T>(&mut self, f: F) -> Option<&mut T> { + unsafe { + self.cleanup_lookup::>(None) + .map(|cln| { + let item = cln.as_ref().data as *mut Item; + &mut (*item).value + }) + .or_else(|| { + self.allocate_with_cleanup(|| Item { value: f() }) + .map(|mut ptr| &mut ptr.as_mut().value) + }) + } + } + + /// Gets the unique value of type `T` from the memory pool. + /// + /// This value must have been previously allocated with [`Pool::get_or_add_unique`]. + /// + /// Returns a reference to the value if found, or `None` if not found. + pub fn get_unique(&self) -> Option<&T> { + unsafe { + self.cleanup_lookup::>(None).map(|cln| { + let item = cln.as_ref().data as *const Item; + &(*item).value + }) + } + } + + /// Gets a mutable reference to the unique value of type `T` from the memory pool. + /// + /// This value must have been previously allocated with [`Pool::get_or_add_unique`]. + /// + /// Returns a mutable reference to the value if found, or `None` if not found. + pub fn get_unique_mut(&mut self) -> Option<&mut T> { + unsafe { + self.cleanup_lookup::>(None).map(|cln| { + let item = cln.as_ref().data as *mut Item; + &mut (*item).value + }) + } + } + + /// Runs the cleanup handler for a value and removes it. + /// + /// Returns `Some(())` if the value was successfully removed, + /// or `None` if the value was not found. + /// + /// # Safety + /// The caller must ensure that `value` is a valid pointer to a value that has an + /// associated cleanup handler in the pool. + pub unsafe fn remove(&self, value: *const T) -> Option<()> { + unsafe { self.remove_cleanup(Some(value)) } + } + + /// Runs the cleanup handler for a unique value and removes it. + /// + /// This value must have been previously allocated with [`Pool::get_or_add_unique`]. + /// + /// Returns `Some(())` if the value was successfully removed, + /// or `None` if the value was not found. + pub fn remove_unique(&self) -> Option<()> { + unsafe { self.remove_cleanup::>(None) } + } + /// Resizes a memory allocation in place if possible. /// /// If resizing is requested for the last allocation in the pool, it may be @@ -338,6 +485,8 @@ impl Pool { /// * `data` - A raw pointer to the value of type `T` to be cleaned up. unsafe extern "C" fn cleanup_type(data: *mut c_void) { unsafe { - ptr::drop_in_place(data as *mut T); + if !data.is_null() { + ptr::drop_in_place(data as *mut T); + } } } diff --git a/src/http/async_request.rs b/src/http/async_request.rs new file mode 100644 index 00000000..9dc587fc --- /dev/null +++ b/src/http/async_request.rs @@ -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; +} + +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, "async handler: Context creation failed") + } + AsyncHandlerError::NoAsyncLauncher => { + write!(f, "async handler: No async launcher available") + } + AsyncHandlerError::ContextDeletionFailed => { + write!(f, "async handler: Context deletion failed") + } + } + } +} + +#[derive(Default)] +struct AsyncRequestContext { + launcher: Option>, +} + +impl HttpRequestHandler for AH +where + AH: AsyncHandler + 'static, +{ + const PHASE: HttpPhase = async_phase(AH::PHASE); + type Output = Result; + + 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::(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("async handler: task should be present"); + let rc = task + .now_or_never() + .expect("async handler: task should be ready"); + ngx_log_debug_http!(request, "async handler: task joined; rc = {}", rc); + pool.remove_unique::() + .ok_or(AsyncHandlerError::ContextDeletionFailed)?; + Ok(rc) + } + Some(_) => { + ngx_log_debug_http!(request, "async handler: running"); + Ok(nginx_sys::NGX_AGAIN as _) + } + } + } +} + +pin_project! { + struct HandlerFuture + where + Fut: Future, + { + #[pin] + worker_fut: Fut, + request: *const ngx_http_request_t, + } +} + +fn handler_future(request: *mut ngx_http_request_t) -> impl Future +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 Future for HandlerFuture +where + Fut: Future, +{ + type Output = ngx_int_t; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + 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, "handler future: 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, "handler future: ready"); + Poll::Ready(rc) + } + } + } +} diff --git a/src/http/mod.rs b/src/http/mod.rs index 00c329a8..8c8ad4a7 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -1,10 +1,26 @@ +#[cfg(feature = "async")] +mod async_request; +// #[cfg(feature = "async")] +// mod async_subrequest; + mod conf; mod module; mod request; +mod request_context; mod status; mod upstream; +/// HTTP subrequest builder and handler. +#[cfg(feature = "alloc")] +pub mod subrequest; + +#[cfg(feature = "async")] +pub use async_request::*; +// #[cfg(feature = "async")] +// pub use async_subrequest::*; + pub use conf::*; pub use module::*; pub use request::*; +pub use request_context::*; pub use status::*; diff --git a/src/http/request.rs b/src/http/request.rs index eb1bd5ac..9b17d5df 100644 --- a/src/http/request.rs +++ b/src/http/request.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 IntoHandlerStatus for Result +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,12 +216,46 @@ impl Request { unsafe { &mut *r.cast::() } } + /// 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::() } + } + /// Is this the main request (as opposed to a subrequest)? pub fn is_main(&self) -> bool { let main = self.0.main.cast(); core::ptr::eq(self, main) } + /// Get a mutable reference to the main request. + /// + /// If this is already the main request, returns `self`; otherwise returns + /// a mutable reference to the associated main request. + pub fn get_main_mut(&mut self) -> &mut Request { + if self.is_main() { + self + } else { + unsafe { Request::from_ngx_http_request(self.0.main) } + } + } + + /// Get a reference to the main request. + /// + /// If this is already the main request, returns `self`; otherwise returns + /// a reference to the associated main request. + pub fn get_main(&self) -> &Request { + if self.is_main() { + self + } else { + unsafe { Request::from_const_ngx_http_request(self.0.main) } + } + } + /// Request pool. pub fn pool(&self) -> Pool { // SAFETY: This request is allocated from `pool`, thus must be a valid pool. @@ -249,6 +303,14 @@ impl Request { unsafe { ctx.as_ref() } } + /// Get mutable Module context + pub fn get_module_ctx_mut(&mut self, module: &ngx_module_t) -> Option<&mut T> { + let ctx = self.get_module_ctx_ptr(module).cast::(); + // SAFETY: ctx is either NULL or allocated with ngx_p(c)alloc and + // explicitly initialized by the module + unsafe { ctx.as_mut() } + } + /// Sets the value as the module's context. /// /// See @@ -293,6 +355,15 @@ impl Request { } } + /// Get HTTP status of response. + pub fn get_status(&self) -> HTTPStatus { + self.0 + .headers_out + .status + .try_into() + .unwrap_or(HTTPStatus(0)) + } + /// Set HTTP status of response. pub fn set_status(&mut self, status: HTTPStatus) { self.0.headers_out.status = status.into(); @@ -362,6 +433,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"); diff --git a/src/http/request_context.rs b/src/http/request_context.rs new file mode 100644 index 00000000..ffbd4bea --- /dev/null +++ b/src/http/request_context.rs @@ -0,0 +1,40 @@ +use crate::http::{HttpModule, Request}; +use crate::ngx_log_debug_http; + +/// A trait for managing request-specific context data. +pub trait RequestContext: Sized { + /// Creates a new context and associates it with the given request. + /// No check is performed to see if a context already exists. + fn create(request: &mut Request, f: F) -> Option<&mut Self> + where + F: FnOnce() -> Self, + { + let ctx_ref = unsafe { request.pool().allocate_with_cleanup(f)?.as_mut() }; + request.set_module_ctx(ctx_ref as *mut _ as _, Module::module()); + Some(ctx_ref) + } + + /// Removes the context associated with the given request. + fn remove(request: &mut Request) { + if let Some(ctx_ptr) = request.get_module_ctx::(Module::module()) { + unsafe { request.pool().remove(ctx_ptr as *const Self) }; + request.set_module_ctx(core::ptr::null_mut(), Module::module()); + ngx_log_debug_http!(request, "RequestContext removed from request"); + } + } + + /// Retrieves an immutable reference to the context associated with the given request. + fn get(request: &Request) -> Option<&Self> { + request.get_module_ctx::(Module::module()) + } + + /// Retrieves a mutable reference to the context associated with the given request. + fn get_mut(request: &mut Request) -> Option<&mut Self> { + request.get_module_ctx_mut::(Module::module()) + } + + /// Checks if a context is associated with the given request. + fn exists(request: &Request) -> bool { + request.get_module_ctx::(Module::module()).is_some() + } +} diff --git a/src/http/status.rs b/src/http/status.rs index a545f65b..29ba5c12 100644 --- a/src/http/status.rs +++ b/src/http/status.rs @@ -4,7 +4,7 @@ use crate::core::Status; use crate::ffi::*; /// Represents an HTTP status code. -#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct HTTPStatus(pub ngx_uint_t); /// A possible error value when converting a `HTTPStatus` from a `u16` or `&str` @@ -12,15 +12,7 @@ pub struct HTTPStatus(pub ngx_uint_t); /// This error indicates that the supplied input was not a valid number, was less /// than 100, or was greater than 599. #[derive(Debug)] -pub struct InvalidHTTPStatusCode { - _priv: (), -} - -impl InvalidHTTPStatusCode { - fn new() -> InvalidHTTPStatusCode { - InvalidHTTPStatusCode { _priv: () } - } -} +pub struct InvalidHTTPStatusCode; impl fmt::Display for InvalidHTTPStatusCode { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { @@ -48,35 +40,42 @@ impl From for ngx_uint_t { } } -impl fmt::Debug for HTTPStatus { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - fmt::Debug::fmt(&self.0, f) - } -} +impl TryFrom for HTTPStatus { + type Error = InvalidHTTPStatusCode; -impl HTTPStatus { - /// Convets a u16 to a status code. #[inline] - pub fn from_u16(src: u16) -> Result { - if !(100..600).contains(&src) { - return Err(InvalidHTTPStatusCode::new()); + fn try_from(value: usize) -> Result { + if !(100..600).contains(&value) { + return Err(InvalidHTTPStatusCode); } + Ok(HTTPStatus(value)) + } +} + +impl TryFrom for HTTPStatus { + type Error = InvalidHTTPStatusCode; - Ok(HTTPStatus(src.into())) + #[inline] + fn try_from(value: isize) -> Result { + let value: usize = value.try_into().map_err(|_| InvalidHTTPStatusCode)?; + Self::try_from(value) } +} + +impl TryFrom<&[u8]> for HTTPStatus { + type Error = InvalidHTTPStatusCode; - /// Converts a &[u8] to a status code. - pub fn from_bytes(src: &[u8]) -> Result { - if src.len() != 3 { - return Err(InvalidHTTPStatusCode::new()); + fn try_from(value: &[u8]) -> Result { + if value.len() != 3 { + return Err(InvalidHTTPStatusCode); } - let a = src[0].wrapping_sub(b'0') as u16; - let b = src[1].wrapping_sub(b'0') as u16; - let c = src[2].wrapping_sub(b'0') as u16; + let a = value[0].wrapping_sub(b'0') as u16; + let b = value[1].wrapping_sub(b'0') as u16; + let c = value[2].wrapping_sub(b'0') as u16; if a == 0 || a > 5 || b > 9 || c > 9 { - return Err(InvalidHTTPStatusCode::new()); + return Err(InvalidHTTPStatusCode); } let status = (a * 100) + (b * 10) + c; diff --git a/src/http/subrequest.rs b/src/http/subrequest.rs new file mode 100644 index 00000000..fb6226bd --- /dev/null +++ b/src/http/subrequest.rs @@ -0,0 +1,343 @@ +use core::ffi::c_void; +use core::fmt::Display; +use core::ptr; + +use alloc::string::{String, ToString}; +use nginx_sys::{ngx_http_post_subrequest_t, ngx_http_request_t, ngx_int_t, ngx_str_t, ngx_uint_t}; + +use crate::{ + core::Pool, + http::{IntoHandlerStatus, Request}, + ngx_log_debug_http, +}; + +#[cfg(feature = "async")] +pub use _async::*; + +/// A builder for creating subrequests. +pub struct SubRequestBuilder { + pool: Pool, + uri: ngx_str_t, + args: Option, + flags: ngx_uint_t, +} + +/// An error type for subrequest operations. +#[derive(Debug)] +pub enum SubRequestError { + /// Indicates that the subrequest allocation failed. + RequestAllocFailed, + /// Indicates that the post subrequest allocation failed. + PostRequestAllocFailed, + /// Indicates that the URI allocation failed. + UriAllocFailed, + /// Indicates that the arguments allocation failed. + ArgsAllocFailed, + /// Indicates that the subrequest creation failed. + CreationFailed, + /// Indicates that the subrequest modification failed. + ModificationFailed(String), +} + +impl Display for SubRequestError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + SubRequestError::RequestAllocFailed => { + write!(f, "subrequest: allocation failed") + } + SubRequestError::PostRequestAllocFailed => { + write!(f, "subrequest: handler allocation failed") + } + SubRequestError::UriAllocFailed => { + write!(f, "subrequest: URI allocation failed") + } + SubRequestError::ArgsAllocFailed => { + write!(f, "subrequest: Arguments allocation failed") + } + SubRequestError::CreationFailed => { + write!(f, "subrequest: creation failed") + } + SubRequestError::ModificationFailed(msg) => { + write!(f, "subrequest: modification failed: {}", msg) + } + } + } +} + +impl SubRequestBuilder { + /// Creates a new `SubRequestBuilder` with the specified URI. + /// The URI is allocated from the provided pool. If the allocation fails, an error is returned. + /// The Pool lifetime must be not shorter than the request which will be used + /// to create the subrequest. + pub fn new(pool: Pool, uri: &str) -> Result { + let uri = unsafe { ngx_str_t::from_bytes(pool.as_ptr(), uri.as_bytes()) } + .ok_or(SubRequestError::UriAllocFailed)?; + Ok(Self { + pool, + uri, + args: None, + flags: 0, + }) + } + + /// Sets the arguments for the subrequest. + pub fn args(mut self, args: &str) -> Result { + let args = unsafe { ngx_str_t::from_bytes(self.pool.as_ptr(), args.as_bytes()) } + .ok_or(SubRequestError::ArgsAllocFailed)?; + self.args = Some(args); + Ok(self) + } + + /// Sets the subrequest to be in-memory. + pub fn in_memory(mut self) -> Self { + self.flags |= nginx_sys::NGX_HTTP_SUBREQUEST_IN_MEMORY as ngx_uint_t; + self + } + + /// Sets the subrequest to be waited. + /// It is supposed to provide some handler to handle the subrequest completion, + /// otherwise it will be waited without any processing + /// (see [`SubRequestBuilder::build_ext`] for details). + pub fn waited(mut self) -> Self { + self.flags |= nginx_sys::NGX_HTTP_SUBREQUEST_WAITED as ngx_uint_t; + self + } + + /// Sets the subrequest to be a background request. + pub fn background(mut self) -> Self { + self.flags |= nginx_sys::NGX_HTTP_SUBREQUEST_BACKGROUND as ngx_uint_t; + self + } + + /// Builds and initiates the subrequest. + /// This method allows for an optional modifier function to modify the subrequest + /// created by `ngx_http_subrequest()` before it is initiated, + /// and an optional handler function to handle the subrequest's completion. + pub fn build_ext( + mut self, + request: &mut Request, + modifier: Option, + handler: Option, + ) -> Result<(), SubRequestError> + where + M: FnOnce(&mut Request) -> Result<(), E>, + E: Display, + H: FnOnce(&mut Request, ngx_int_t) -> O, + O: IntoHandlerStatus, + { + let sr_args_ptr = self + .args + .as_mut() + .map_or(ptr::null_mut(), |args| args as *mut ngx_str_t); + + let psr_ptr: *mut ngx_http_post_subrequest_t = if handler.is_some() { + let ctx = unsafe { + self.pool + .allocate_with_cleanup(|| handler) + .ok_or(SubRequestError::RequestAllocFailed) + }?; + + let psr = unsafe { + self.pool + .allocate_with_cleanup(|| ngx_http_post_subrequest_t { + handler: Some(sr_handler::), + data: ctx.as_ptr() as _, + }) + .ok_or(SubRequestError::PostRequestAllocFailed) + }?; + psr.as_ptr() as _ + } else { + ptr::null_mut() + }; + + let mut sr_ptr: *mut ngx_http_request_t = core::ptr::null_mut(); + + let rc = unsafe { + nginx_sys::ngx_http_subrequest( + request.as_mut() as *mut _ as _, + &raw mut self.uri, + sr_args_ptr, + &raw mut sr_ptr, + psr_ptr, + self.flags as ngx_uint_t, + ) + }; + if rc != nginx_sys::NGX_OK as _ { + return Err(SubRequestError::CreationFailed); + } + + if let Some(modifier) = modifier { + let sr = unsafe { Request::from_ngx_http_request(sr_ptr) }; + modifier(sr).map_err(|e| SubRequestError::ModificationFailed(e.to_string())) + } else { + Ok(()) + } + } + + /// Builds and initiates the subrequest. + /// This is a simplified version of `build_ext` that requires a handler + /// and does not allow for subrequest modification. + pub fn build(self, request: &mut Request, handler: H) -> Result<(), SubRequestError> + where + H: FnOnce(&mut Request, ngx_int_t) -> O, + O: IntoHandlerStatus, + { + self.build_ext(request, SR_MODIFIER_NO_OP, Some(handler)) + } +} + +type SimpleSubRequestModifier = Option Result<(), core::convert::Infallible>>; +type SimpleSubRequestHandler = Option ngx_int_t>; +/// A no-op modifier function for subrequests. +pub const SR_MODIFIER_NO_OP: SimpleSubRequestModifier = None; +/// A no-op handler function for subrequests. +pub const SR_HANDLER_NO_OP: SimpleSubRequestHandler = None; + +extern "C" fn sr_handler( + r: *mut ngx_http_request_t, + data: *mut c_void, + rc: ngx_int_t, +) -> ngx_int_t +where + H: FnOnce(&mut Request, ngx_int_t) -> O, + O: IntoHandlerStatus, +{ + let request = unsafe { Request::from_ngx_http_request(r) }; + ngx_log_debug_http!(request, "subrequest handler called with rc: {rc}"); + if let Some(handler) = unsafe { &mut *(data as *mut Option) }.take() { + (handler)(request, rc).into_handler_status(request) + } else { + rc + } +} + +#[cfg(feature = "async")] +impl SubRequestBuilder { + /// Builds and runs the subrequest asynchronously. + pub async fn build_async<'r, M, E, AH>( + mut self, + request: &'r mut Request, + modifier: Option, + handler: AH, + ) -> Result + where + M: FnOnce(&mut Request) -> Result<(), E>, + E: Display, + AH: FnMut(&'r mut Request, ngx_int_t) -> ngx_int_t + Unpin, + { + let sr_args_ptr = self + .args + .as_mut() + .map_or(ptr::null_mut(), |args| args as *mut ngx_str_t); + + let mut ctx = core::pin::pin!(AsyncSubRequest::::new(handler)); + + let mut psr = core::pin::pin!(ngx_http_post_subrequest_t { + handler: Some(AsyncSubRequest::::sr_handler), + data: ctx.as_mut().get_mut() as *mut _ as _, + }); + + let mut sr_ptr: *mut ngx_http_request_t = core::ptr::null_mut(); + + let rc = unsafe { + nginx_sys::ngx_http_subrequest( + request.as_mut() as *mut _ as _, + &raw mut self.uri, + sr_args_ptr, + &raw mut sr_ptr, + psr.as_mut().get_mut() as _, + self.flags as ngx_uint_t, + ) + }; + if rc != nginx_sys::NGX_OK as _ { + return Err(SubRequestError::CreationFailed); + } + + if let Some(modifier) = modifier { + let sr = unsafe { Request::from_ngx_http_request(sr_ptr) }; + modifier(sr).map_err(|e| SubRequestError::ModificationFailed(e.to_string()))?; + } + + ctx.await + } +} + +#[cfg(feature = "async")] +mod _async { + + use futures_util::task::noop_waker; + + use super::*; + + use core::pin::Pin; + use core::task::Waker; + + use crate::ngx_log_debug_http; + + /// An asynchronous subrequest structure. + pub struct AsyncSubRequest<'sr, H> + where + H: FnMut(&'sr mut Request, ngx_int_t) -> ngx_int_t + Unpin, + { + _phantom: core::marker::PhantomData<&'sr ()>, + handler: Option, + waker: Waker, + rc: Option, + } + + impl<'sr, H> AsyncSubRequest<'sr, H> + where + H: FnMut(&'sr mut Request, ngx_int_t) -> ngx_int_t + Unpin, + { + pub(super) fn new(handler: H) -> Self { + Self { + _phantom: core::marker::PhantomData, + handler: Some(handler), + waker: noop_waker(), + rc: None, + } + } + + pub(super) extern "C" fn sr_handler( + r: *mut ngx_http_request_t, + data: *mut c_void, + mut rc: ngx_int_t, + ) -> ngx_int_t { + let request = unsafe { Request::from_ngx_http_request(r) }; + ngx_log_debug_http!( + request, + "async subrequest done rc:{} s:{}", + rc, + request.get_status().0 + ); + + let this = unsafe { &mut *(data as *mut Self) }; + this.rc = Some(rc); + if let Some(mut handler) = this.handler.take() { + rc = handler(request, rc); + } + this.waker.wake_by_ref(); + rc + } + } + + impl<'sr, H> core::future::Future for AsyncSubRequest<'sr, H> + where + H: FnMut(&'sr mut Request, ngx_int_t) -> ngx_int_t + Unpin, + { + type Output = Result; + + fn poll( + mut self: Pin<&mut Self>, + cx: &mut core::task::Context<'_>, + ) -> core::task::Poll { + match self.rc { + None => { + self.waker.clone_from(cx.waker()); + core::task::Poll::Pending + } + Some(rc) => core::task::Poll::Ready(Ok(rc)), + } + } + } +} diff --git a/src/log.rs b/src/log.rs index c3ebe7ee..47a1479e 100644 --- a/src/log.rs +++ b/src/log.rs @@ -153,7 +153,7 @@ macro_rules! ngx_log_debug { #[macro_export] macro_rules! ngx_log_debug_http { ( $request:expr, $($arg:tt)+ ) => { - let log = unsafe { (*$request.connection()).log }; + let log = $request.log(); $crate::ngx_log_debug!(mask: $crate::log::DebugMask::Http, log, $($arg)+); } }