Add async_eventfd feature to optionally optimize notify method on epoll

Author: pschyska

Cargo.toml

@@ -54,6 +54,8 @@ async = [
     "dep:pin-project-lite",
     "dep:crossbeam-channel",
 ]
+# Optimize async notification with eventfd, requires nginx epoll event module
+async_eventfd = ["async"]
 # Provides APIs that require allocations via the `alloc` crate.
 alloc = ["allocator-api2/alloc"]
 # Enables serialization support for some of the provided and re-exported types.

examples/Cargo.toml

@@ -75,3 +75,4 @@ default = ["export-modules", "ngx/vendored"]
 export-modules = []
 linux = []
 async = ["ngx/async"]
+async_eventfd = ["async", "ngx/async_eventfd"]

examples/async.rs

@@ -7,16 +7,6 @@ use hyper_util::rt::TokioIo;
 use nginx_sys::{ngx_http_core_loc_conf_t, NGX_LOG_ERR};
 use ngx::async_::resolver::Resolver;
 use ngx::async_::{spawn, Task};
-use std::cell::RefCell;
-use std::ffi::{c_char, c_void};
-use std::future::Future;
-use std::pin::Pin;
-use std::ptr::{addr_of, addr_of_mut, NonNull};
-use std::sync::atomic::{AtomicPtr, Ordering};
-use std::task::Poll;
-use std::time::Instant;
-use tokio::net::TcpStream;
-
 use ngx::core::{self, Pool, Status};
 use ngx::ffi::{
     ngx_array_push, ngx_command_t, ngx_conf_t, ngx_connection_t, ngx_http_handler_pt,
@@ -29,6 +19,17 @@ use ngx::http::{HttpModuleLocationConf, HttpModuleMainConf, NgxHttpCoreModule};
 use ngx::{
     http_request_handler, ngx_conf_log_error, ngx_log_debug_http, ngx_log_error, ngx_string,
 };
+use std::cell::RefCell;
+use std::ffi::{c_char, c_void};
+use std::future::Future;
+use std::pin::Pin;
+use std::ptr::{addr_of, addr_of_mut, NonNull};
+use std::sync::atomic::{AtomicPtr, Ordering};
+use std::task::Poll;
+use std::time::Instant;
+use tokio::net::TcpStream;
+#[cfg(feature = "async_eventfd")]
+use {nginx_sys::ngx_cycle_t, ngx::async_::async_init};
 
 struct Module;
 
@@ -98,6 +99,10 @@ pub static mut ngx_http_async_module: ngx_module_t = ngx_module_t {
     ctx: std::ptr::addr_of!(NGX_HTTP_ASYNC_MODULE_CTX) as _,
     commands: unsafe { &NGX_HTTP_ASYNC_COMMANDS[0] as *const _ as *mut _ },
     type_: NGX_HTTP_MODULE as _,
+    #[cfg(not(feature = "async_eventfd"))]
+    init_process: None,
+    #[cfg(feature = "async_eventfd")]
+    init_process: Some(ngx_http_async_init),
     ..ngx_module_t::default()
 };
 
@@ -166,7 +171,7 @@ async fn resolve_something(
         .expect("resolution");
 
     (
-        format!("X-Resolve-Time"),
+        "X-Resolve-Time".to_string(),
         start.elapsed().as_millis().to_string(),
     )
 }
@@ -188,7 +193,7 @@ async fn reqwest_something() -> (String, String) {
 async fn hyper_something() -> (String, String) {
     let start = Instant::now();
     // see https://hyper.rs/guides/1/client/basic/
-    let url = "http://httpbin.org/ip".parse::<hyper::Uri>().expect("uri");
+    let url = "https://example.com".parse::<hyper::Uri>().expect("uri");
     let host = url.host().expect("uri has no host");
     let port = url.port_u16().unwrap_or(80);
 
@@ -331,3 +336,9 @@ extern "C" fn ngx_http_async_commands_set_enable(
 
     ngx::core::NGX_CONF_OK
 }
+
+#[cfg(feature = "async_eventfd")]
+extern "C" fn ngx_http_async_init(_cycle: *mut ngx_cycle_t) -> ngx_int_t {
+    async_init();
+    Status::NGX_OK.into()
+}

nginx-sys/build/main.rs

@@ -20,6 +20,7 @@ const NGX_CONF_FEATURES: &[&str] = &[
     "compat",
     "debug",
     "have_epollrdhup",
+    "have_eventfd",
     "have_file_aio",
     "have_kqueue",
     "have_memalign",

src/async_/mod.rs

@@ -1,6 +1,8 @@
 //! Async runtime and set of utilities on top of the NGINX event loop.
 pub use self::sleep::{sleep, Sleep};
 pub use self::spawn::{spawn, Task};
+#[cfg(feature = "async_eventfd")]
+pub use self::spawn::async_init;
 
 pub mod resolver;
 

src/async_/spawn.rs

@@ -1,21 +1,135 @@
 extern crate std;
 
-use core::ffi::c_int;
+use core::mem;
 use core::sync::atomic::{AtomicI64, Ordering};
-use core::{mem, ptr};
-use std::sync::OnceLock;
 
 use core::future::Future;
+use std::sync::OnceLock;
 
-use alloc::boxed::Box;
 pub use async_task::Task;
 use async_task::{Runnable, ScheduleInfo, WithInfo};
 use crossbeam_channel::{unbounded, Receiver, Sender};
-use nginx_sys::{kill, ngx_event_t, ngx_post_event, ngx_posted_next_events, ngx_thread_tid, SIGIO};
+use nginx_sys::{ngx_event_t, ngx_thread_tid};
 
 use crate::log::ngx_cycle_log;
 use crate::ngx_log_debug;
 
+static mut NOTIFY_EVENT: ngx_event_t = unsafe { mem::zeroed() };
+
+#[cfg(not(feature = "async_eventfd"))]
+mod default {
+    extern crate std;
+
+    use nginx_sys::{kill, ngx_log_t, ngx_post_event, ngx_posted_next_events, SIGIO};
+
+    use super::async_handler;
+    use super::NOTIFY_EVENT;
+    use crate::log::ngx_cycle_log;
+    use crate::ngx_log_debug;
+
+    pub(crate) fn notify() {
+        let log: *mut ngx_log_t = ngx_cycle_log().as_ptr();
+        let rc = unsafe {
+            NOTIFY_EVENT.log = log;
+            NOTIFY_EVENT.handler = Some(async_handler);
+
+            ngx_post_event(&raw mut NOTIFY_EVENT, &raw mut ngx_posted_next_events);
+            kill(
+                std::process::id().try_into().unwrap(),
+                SIGIO.try_into().unwrap(),
+            )
+        };
+        if rc != 0 {
+            panic!("kill: rc={rc}");
+        }
+
+        ngx_log_debug!(log, "async: notified (SIGIO)");
+    }
+}
+#[cfg(not(feature = "async_eventfd"))]
+use default::*;
+
+#[cfg(feature = "async_eventfd")]
+mod eventfd {
+    extern crate std;
+
+    use core::mem;
+
+    use nginx_sys::{
+        eventfd, ngx_connection_t, ngx_cycle, ngx_event_actions, ngx_event_t, ngx_int_t, ngx_log_t,
+        EFD_CLOEXEC, EFD_NONBLOCK, EPOLL_EVENTS_EPOLLET, EPOLL_EVENTS_EPOLLIN,
+        EPOLL_EVENTS_EPOLLRDHUP, NGX_ERROR, NGX_OK,
+    };
+    use std::{fs::File, io::Write, os::fd::FromRawFd, sync::OnceLock};
+
+    use super::async_handler;
+    use super::NOTIFY_EVENT;
+    use crate::log::ngx_cycle_log;
+    use crate::ngx_log_debug;
+
+    #[cfg(not(ngx_feature = "have_eventfd"))]
+    compile_error!("feature async_eventfd requires eventfd(), NGX_HAVE_EVENTFD");
+
+    static mut NOTIFY_CONN: ngx_connection_t = unsafe { mem::zeroed() };
+    static EVENT_FILE: OnceLock<File> = OnceLock::new();
+
+    pub(crate) fn notify() {
+        let w = EVENT_FILE
+            .get()
+            .expect("EVENT_FILE")
+            .write(&1u64.to_ne_bytes())
+            .expect("write");
+        if w != 8 {
+            panic!("eventfd: wrote {w}, expected 8");
+        }
+
+        ngx_log_debug!(ngx_cycle_log().as_ptr(), "async: notified (eventfd)");
+    }
+
+    static mut _DUMMY_WRITE_EVENT: ngx_event_t = unsafe { mem::zeroed() };
+
+    extern "C" fn _dummy_write_handler(_ev: *mut ngx_event_t) {}
+
+    /// intialize async_eventfd, needs to be called once in init_process, before spawn() is used
+    pub fn async_init() -> ngx_int_t {
+        let fd = unsafe { eventfd(0, (EFD_NONBLOCK | EFD_CLOEXEC).try_into().unwrap()) };
+        if fd == -1 {
+            return NGX_ERROR as isize;
+        }
+
+        let log: *mut ngx_log_t = unsafe { ngx_cycle.read().log };
+
+        ngx_log_debug!(log, "async: eventfd {fd}");
+
+        unsafe {
+            EVENT_FILE.set(File::from_raw_fd(fd)).expect("EVENT_FILE");
+
+            NOTIFY_EVENT.handler = Some(async_handler);
+            NOTIFY_EVENT.log = log;
+            #[allow(static_mut_refs)]
+            NOTIFY_EVENT.set_active(1);
+            NOTIFY_EVENT.data = (&raw mut NOTIFY_CONN).cast();
+
+            _DUMMY_WRITE_EVENT.handler = Some(_dummy_write_handler);
+
+            NOTIFY_CONN.fd = fd;
+            NOTIFY_CONN.read = &raw mut NOTIFY_EVENT;
+            NOTIFY_CONN.write = &raw mut _DUMMY_WRITE_EVENT;
+            NOTIFY_CONN.log = log;
+
+            ngx_event_actions.add.unwrap()(
+                &raw mut NOTIFY_EVENT,
+                (EPOLL_EVENTS_EPOLLIN | EPOLL_EVENTS_EPOLLRDHUP) as isize,
+                EPOLL_EVENTS_EPOLLET as usize,
+            );
+        }
+
+        NGX_OK as ngx_int_t
+    }
+}
+#[cfg(feature = "async_eventfd")]
+pub use eventfd::*;
+
 static MAIN_TID: AtomicI64 = AtomicI64::new(-1);
 
 #[inline]
@@ -25,34 +139,21 @@ fn on_event_thread() -> bool {
     main_tid == tid
 }
 
-extern "C" fn async_handler(ev: *mut ngx_event_t) {
+extern "C" fn async_handler(_ev: *mut ngx_event_t) {
     // initialize MAIN_TID on first execution
     let tid = unsafe { ngx_thread_tid().into() };
     let _ = MAIN_TID.compare_exchange(-1, tid, Ordering::Relaxed, Ordering::Relaxed);
     let scheduler = scheduler();
+
+    if scheduler.rx.is_empty() {
+        return;
+    }
     let mut cnt = 0;
     while let Ok(r) = scheduler.rx.try_recv() {
         r.run();
         cnt += 1;
     }
-    ngx_log_debug!(
-        unsafe { (*ev).log },
-        "async: processed {cnt} items"
-    );
-
-    unsafe {
-        drop(Box::from_raw(ev));
-    }
-}
-
-fn notify() -> c_int {
-    ngx_log_debug!(ngx_cycle_log().as_ptr(), "async: notify via SIGIO");
-    unsafe {
-        kill(
-            std::process::id().try_into().unwrap(),
-            SIGIO.try_into().unwrap(),
-        )
-    }
+    ngx_log_debug!(ngx_cycle_log().as_ptr(), "async: processed {cnt} items");
 }
 
 struct Scheduler {
@@ -69,28 +170,17 @@ impl Scheduler {
     fn schedule(&self, runnable: Runnable, info: ScheduleInfo) {
         let oet = on_event_thread();
         // If we are on the event loop thread it's safe to simply run the Runnable, otherwise we
-        // enqueue the Runnable, post our event, and SIGIO to interrupt epoll. The event handler
-        // then runs the Runnable on the event loop thread.
+        // enqueue the Runnable, post our event, and notify. The event handler then runs the
+        // Runnable on the event loop thread.
         //
         // If woken_while_running, it indicates that a task has yielded itself to the Scheduler.
-        // Force round-trip via queue to limit reentrancy (skipping SIGIO).
+        // Force round-trip via queue to limit reentrancy.
         if oet && !info.woken_while_running {
             runnable.run();
         } else {
             self.tx.send(runnable).expect("send");
-            unsafe {
-                let event: *mut ngx_event_t = Box::into_raw(Box::new(mem::zeroed()));
-                (*event).handler = Some(async_handler);
-                (*event).log = ngx_cycle_log().as_ptr();
-                ngx_post_event(event, ptr::addr_of_mut!(ngx_posted_next_events));
-            }
-
-            if !oet {
-                let rc = notify();
-                if rc != 0 {
-                    panic!("kill: {rc}")
-                }
-            }
+
+            notify();
         }
     }
 }