Create an abstraction for the tokio runtime to allow for different ownership configurations.

Author: jmaygarden

Cargo.toml

@@ -43,6 +43,7 @@ tokio = { version = "1", optional = true, features = [
     "macros",
     "net",
     "rt",
+    "rt-multi-thread",
     "signal",
     "sync",
     "time",

examples/hello/src/main.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 pub mod ffi;
 
-use libevent::{EventCallbackCtx, EventCallbackFlags, EvutilSocket};
+use libevent::{tokio_backend::Runtime, EventCallbackCtx, EventCallbackFlags, EvutilSocket};
 
 extern "C" fn hello_callback(
     _fd: EvutilSocket,
@@ -15,19 +15,21 @@ extern "C" fn hello_callback(
 
 #[cfg(feature = "tokio_backend")]
 fn inject_tokio(base: &Base) {
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .expect("failed to build a tokio runtime");
-
-    runtime.spawn(async {
-        loop {
-            tokio::time::sleep(Duration::from_secs(5)).await;
-            println!("'Hello, world' from a tokio task!");
-        }
-    });
+    let runtime =
+        libevent::tokio_backend::TokioRuntime::new().expect("failed to build a tokio runtime");
+
+    {
+        let _guard = runtime.enter();
+
+        tokio::spawn(async {
+            loop {
+                tokio::time::sleep(Duration::from_secs(3)).await;
+                println!("'Hello, world' from a tokio task!");
+            }
+        });
+    }
 
-    base.inject_tokio(runtime);
+    base.inject_tokio(Box::new(runtime));
 }
 
 fn main() {

libevent-sys/src/lib.rs

@@ -10,9 +10,11 @@
 #![allow(clippy::redundant_static_lifetimes)]
 #![allow(clippy::too_many_arguments)]
 #![allow(clippy::unreadable_literal)]
+#![allow(deref_nullptr)]
 #![allow(non_camel_case_types)]
 #![allow(non_snake_case)]
 #![allow(non_upper_case_globals)]
+#![allow(unaligned_references)]
 
 include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
 
@@ -23,6 +25,6 @@ mod tests {
     #[test]
     fn constant_access() {
         assert_eq!(EVENT_LOG_MSG, 1);
-        assert_eq!(IPPORT_RESERVED, 1024);
+        assert_eq!(IPV6PORT_RESERVED, 1024);
     }
 }

sample/bench.c

@@ -34,6 +34,7 @@
  */
 
 #include "event2/event-config.h"
+#include "event2/thread.h"
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -173,6 +174,10 @@ main(int argc, char **argv)
 	struct timeval *tv;
 	evutil_socket_t *cp;
 
+#ifndef USE_TOKIO
+	evthread_use_pthreads();
+#endif
+
 	num_pipes = 100;
 	num_active = 1;
 	num_writes = num_pipes;

src/base.rs

@@ -1,14 +1,13 @@
 #![allow(dead_code)]
 
+use super::{event::*, tokio_backend::Runtime};
+use crate::EventCallbackWrapper;
 use bitflags::bitflags;
 use std::io;
 use std::os::raw::{c_int, c_short, c_void};
 use std::ptr::NonNull;
 use std::time::Duration;
 
-use super::event::*;
-use crate::EventCallbackWrapper;
-
 /// A file descriptor in libevent.
 pub type EvutilSocket = c_int;
 
@@ -51,9 +50,9 @@ impl Base {
         }
     }
 
-    /// Replaces the standard libevent backend with tokio
+    /// Replaces the standard libevent backend with an owned tokio runtime
     #[cfg(feature = "tokio_backend")]
-    pub fn inject_tokio(&self, runtime: tokio::runtime::Runtime) {
+    pub fn inject_tokio(&self, runtime: Box<dyn Runtime>) {
         super::tokio_backend::inject_tokio(self.base, runtime)
     }
 

src/lib.rs

@@ -13,7 +13,7 @@ pub use base::{
 };
 
 #[cfg(feature = "tokio_backend")]
-mod tokio_backend;
+pub mod tokio_backend;
 
 /// The context passed into `handle_wrapped_callback`, which handles event-type
 /// specific metadata for trampolining into the user-supplied closure.

src/tokio_backend/api.rs

@@ -1,4 +1,9 @@
-use super::{backend::TokioBackend, io::IoType, BaseWrapper};
+use super::{
+    backend::TokioBackend,
+    io::IoType,
+    runtime::{Runtime, TokioRuntime},
+    BaseWrapper,
+};
 use std::{
     ffi::c_void,
     os::{
@@ -9,40 +14,34 @@ use std::{
     time::Duration,
 };
 
+const EVSEL: libevent_sys::eventop = libevent_sys::eventop {
+    name: "tokio".as_ptr().cast(),
+    init: Some(tokio_backend_init),
+    add: Some(tokio_backend_add),
+    del: Some(tokio_backend_del),
+    dispatch: Some(tokio_backend_dispatch),
+    dealloc: Some(tokio_backend_dealloc),
+    need_reinit: 1,
+    features: libevent_sys::event_method_feature_EV_FEATURE_FDS,
+    fdinfo_len: std::mem::size_of::<RawFd>() as u64,
+};
+const EVSIGSEL: libevent_sys::eventop = libevent_sys::eventop {
+    name: "tokio_signal".as_ptr().cast(),
+    init: None,
+    add: Some(tokio_signal_backend_add),
+    del: Some(tokio_signal_backend_del),
+    dispatch: None,
+    dealloc: None,
+    need_reinit: 0,
+    features: 0,
+    fdinfo_len: 0,
+};
+
 /// Injects a tokio backend with the given runtime into the given libevent instance.
 ///
 /// The libevent instance will already have an initialized backend. This
 /// exisiting backend is deallocated before being replaced.
-///
-/// A tracing-subscriber may also be initialized if the feature is activated
-/// to enable tracing output when linked to a C program.
-pub fn inject_tokio(mut base: NonNull<libevent_sys::event_base>, runtime: tokio::runtime::Runtime) {
-    const EVSEL: libevent_sys::eventop = libevent_sys::eventop {
-        name: "tokio".as_ptr().cast(),
-        init: Some(tokio_backend_init),
-        add: Some(tokio_backend_add),
-        del: Some(tokio_backend_del),
-        dispatch: Some(tokio_backend_dispatch),
-        dealloc: Some(tokio_backend_dealloc),
-        need_reinit: 1,
-        features: libevent_sys::event_method_feature_EV_FEATURE_FDS,
-        fdinfo_len: std::mem::size_of::<RawFd>() as u64,
-    };
-    const EVSIGSEL: libevent_sys::eventop = libevent_sys::eventop {
-        name: "tokio_signal".as_ptr().cast(),
-        init: None,
-        add: Some(tokio_signal_backend_add),
-        del: Some(tokio_signal_backend_del),
-        dispatch: None,
-        dealloc: None,
-        need_reinit: 0,
-        features: 0,
-        fdinfo_len: 0,
-    };
-
-    #[cfg(feature = "tracing_subscriber")]
-    tracing_subscriber::fmt::init();
-
+pub fn inject_tokio(mut base: NonNull<libevent_sys::event_base>, runtime: Box<dyn Runtime>) {
     let backend = Box::new(TokioBackend::new(runtime));
     let base = unsafe { base.as_mut() };
 
@@ -65,16 +64,18 @@ pub unsafe extern "C" fn tokio_event_base_new() -> *mut libevent_sys::event_base
     let base = NonNull::new(libevent_sys::event_base_new());
 
     match base {
-        Some(base) => {
-            let runtime = tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .expect("failed to build a tokio runtime");
+        Some(base) => match TokioRuntime::new() {
+            Ok(runtime) => {
+                inject_tokio(base, Box::new(runtime));
 
-            inject_tokio(base, runtime);
+                base.as_ptr()
+            }
+            Err(error) => {
+                tracing::error!(?error, "failed to create a new Tokio runtime");
 
-            base.as_ptr()
-        }
+                std::ptr::null_mut()
+            }
+        },
         None => std::ptr::null_mut(),
     }
 }

src/tokio_backend/backend.rs

@@ -1,16 +1,17 @@
 use super::{
     io::{IoMap, IoType},
+    runtime::Runtime,
     signal::SignalMap,
     BaseWrapper,
 };
+use crate::tokio_backend::runtime::{JoinFuture, YieldFuture};
 use std::{os::raw::c_int, sync::Arc, time::Duration};
-use tokio::sync::Notify;
+use tokio::{sync::Notify, task::JoinHandle};
 
 /// Implements a libevent backend using a tokio runtime
-#[derive(Debug)]
 pub struct TokioBackend {
-    /// Tokio runtime for driving I/O and signal events
-    runtime: tokio::runtime::Runtime,
+    /// Option tokio runtime to maintain ownership
+    runtime: Box<dyn Runtime>,
     /// Notifies the dispatch loop that it should yield back to libevent
     dispatch_notify: Arc<Notify>,
     /// Map of futures for registered I/O events
@@ -20,8 +21,8 @@ pub struct TokioBackend {
 }
 
 impl TokioBackend {
-    /// Create a new tokio libevent backend using the provided runtime
-    pub fn new(runtime: tokio::runtime::Runtime) -> Self {
+    /// Create a new libevent backend using the provided runtime
+    pub fn new(runtime: Box<dyn Runtime>) -> Self {
         let dispatch_notify = Arc::new(Notify::new());
         let io_map = IoMap::new();
         let signal_map = SignalMap::new();
@@ -62,14 +63,19 @@ impl TokioBackend {
         }
     }
 
+    /// Blocks on the given join handle
+    fn join(&self, join_handle: JoinHandle<()>) {
+        let future = JoinFuture::new(join_handle);
+
+        self.runtime.join(future);
+    }
+
     /// Terminates an active I/O task
     pub fn del_io(&mut self, fd: c_int) -> c_int {
         tracing::debug!(fd, "delete an I/O event");
 
         if let Ok(join_handle) = self.io_map.del(fd) {
-            self.runtime.block_on(async move {
-                let _ = join_handle.await;
-            });
+            self.join(join_handle);
             0
         } else {
             -1
@@ -104,12 +110,10 @@ impl TokioBackend {
 
     /// Terminates an active signal task
     pub fn del_signal(&mut self, signum: c_int) -> c_int {
-        tracing::debug!(signum, "delete an signal event");
+        tracing::debug!(signum, "delete a signal event");
 
         if let Ok(join_handle) = self.signal_map.del(signum) {
-            self.runtime.block_on(async move {
-                let _ = join_handle.await;
-            });
+            self.join(join_handle);
             0
         } else {
             -1
@@ -120,19 +124,26 @@ impl TokioBackend {
     pub fn dispatch(&mut self, _base: *mut libevent_sys::event_base, timeout: Option<Duration>) {
         tracing::trace!(?timeout, "dispatch events");
 
-        let dispatch_notify = self.dispatch_notify.clone();
+        let _guard = self.runtime.enter();
 
-        self.runtime.block_on(async move {
-            if let Some(duration) = timeout {
+        match timeout {
+            Some(duration) => {
                 if duration.is_zero() {
-                    tokio::task::yield_now().await;
-                    tokio::task::yield_now().await;
+                    let future = YieldFuture::default();
+
+                    self.runtime.dispatch_yield(future);
                 } else {
-                    let _ = tokio::time::timeout(duration, dispatch_notify.notified()).await;
+                    let future = self.dispatch_notify.notified();
+                    let future = tokio::time::timeout(duration, future);
+
+                    self.runtime.dispatch_timeout(future);
                 }
-            } else {
-                dispatch_notify.notified().await;
             }
-        })
+            None => {
+                let future = self.dispatch_notify.notified();
+
+                self.runtime.dispatch_notify(future);
+            }
+        }
     }
 }

src/tokio_backend/io.rs

@@ -1,6 +1,10 @@
 use super::BaseWrapper;
 use std::{collections::HashMap, os::unix::prelude::RawFd, sync::Arc};
-use tokio::{io::{unix::AsyncFd, Interest}, sync::Notify, task::JoinHandle};
+use tokio::{
+    io::{unix::AsyncFd, Interest},
+    sync::Notify,
+    task::JoinHandle,
+};
 
 /// Manages adding and removing I/O event tasks
 #[derive(Debug)]

src/tokio_backend/mod.rs

@@ -1,9 +1,11 @@
 mod api;
 mod backend;
 mod io;
+mod runtime;
 mod signal;
 
 pub use api::inject_tokio;
+pub use runtime::{JoinFuture, Runtime, TokioRuntime, YieldFuture};
 
 /// Wrapper to allow sending of raw event_base pointers to tokio tasks.
 ///