Refactor Runtimes to use separate TLS implementations internally.

This will make it easier to implement them using some other TLS
provider as well, without having to duplicate all of our code.

Cargo.lock

@@ -2679,7 +2679,6 @@ name = "tor-circmgr"
 version = "0.0.3"
 dependencies = [
  "async-trait",
- "async_executors",
  "bounded-vec-deque",
  "derive_builder",
  "futures",

crates/tor-circmgr/Cargo.toml

@@ -44,7 +44,6 @@ thiserror = "1"
 weak-table = "0.3.0"
 
 [dev-dependencies]
-async_executors = { version = "0.4", default-features = false, features = [ "tokio_tp" ] }
 futures-await-test = "0.3.0"
 tor-rtmock = { path="../tor-rtmock", version = "0.0.3"}
 tor-guardmgr = { path="../tor-guardmgr", version = "0.0.3", features=["testing"]}

crates/tor-circmgr/src/lib.rs

@@ -438,7 +438,7 @@ mod test {
 
     /// Helper type used to help type inference.
     pub(crate) type OptDummyGuardMgr<'a> =
-        Option<&'a tor_guardmgr::GuardMgr<async_executors::TokioTp>>;
+        Option<&'a tor_guardmgr::GuardMgr<tor_rtcompat::tokio::TokioRuntime>>;
 
     #[test]
     fn get_params() {

crates/tor-rtcompat/src/async_std.rs

 //! Entry points for use with async_std runtimes.
-pub use crate::impls::async_std::create_runtime as create_async_std_runtime;
-use crate::SpawnBlocking;
+pub use crate::impls::async_std::create_runtime as create_runtime_impl;
+use crate::{compound::CompoundRuntime, SpawnBlocking};
 
-/// A [`Runtime`](crate::Runtime) powered by async-std.
-pub use async_executors::AsyncStd as AsyncStdRuntime;
+use crate::impls::async_std::NativeTlsAsyncStd;
+
+use async_executors::AsyncStd;
+
+/// A [`Runtime`](crate::Runtime) powered by `async_std` and `native_tls`.
+#[derive(Clone)]
+pub struct AsyncStdRuntime {
+    /// The actual runtime object.
+    inner: Inner,
+}
+
+/// Implementation type for AsyncStdRuntime.
+type Inner = CompoundRuntime<AsyncStd, AsyncStd, AsyncStd, NativeTlsAsyncStd>;
+
+crate::opaque::implement_opaque_runtime! {
+    AsyncStdRuntime { inner : Inner }
+}
 
 /// Return a new async-std-based [`Runtime`](crate::Runtime).
 ///
@@ -12,7 +27,10 @@ pub use async_executors::AsyncStd as AsyncStdRuntime;
 /// runtime.
 
 pub fn create_runtime() -> std::io::Result<AsyncStdRuntime> {
-    Ok(create_async_std_runtime())
+    let rt = create_runtime_impl();
+    Ok(AsyncStdRuntime {
+        inner: CompoundRuntime::new(rt, rt, rt, NativeTlsAsyncStd::default()),
+    })
 }
 
 /// Try to return an instance of the currently running async_std
@@ -28,6 +46,6 @@ where
     P: FnOnce(AsyncStdRuntime) -> F,
     F: futures::Future<Output = O>,
 {
-    let runtime = create_async_std_runtime();
-    runtime.block_on(func(runtime))
+    let runtime = current_runtime().expect("Couldn't get global async_std runtime?");
+    runtime.clone().block_on(func(runtime))
 }

crates/tor-rtcompat/src/impls/async_std.rs

@@ -205,7 +205,12 @@ impl SpawnBlocking for async_executors::AsyncStd {
     }
 }
 
-impl TlsProvider<TcpStream> for async_executors::AsyncStd {
+/// A TlsProvider that uses native_tls and works with the AsyncStd executor.
+#[derive(Clone, Debug, Default)]
+#[non_exhaustive]
+pub struct NativeTlsAsyncStd {}
+
+impl TlsProvider<TcpStream> for NativeTlsAsyncStd {
     type TlsStream = tls::TlsStream;
     type Connector = tls::TlsConnector;
 

crates/tor-rtcompat/src/impls/tokio.rs

@@ -219,6 +219,29 @@ mod tls {
 
 // ==============================
 
+/// A TlsProvider that uses native_tls and works with the Tokio executor.
+#[derive(Clone, Debug, Default)]
+#[non_exhaustive]
+pub struct NativeTlsTokio {}
+
+impl TlsProvider<net::TcpStream> for NativeTlsTokio {
+    type TlsStream = tls::TlsStream;
+    type Connector = tls::TlsConnector;
+
+    fn tls_connector(&self) -> tls::TlsConnector {
+        let mut builder = native_tls::TlsConnector::builder();
+        // These function names are scary, but they just mean that we
+        // aren't checking whether the signer of this cert
+        // participates in the web PKI, and we aren't checking the
+        // hostname in the cert.
+        builder
+            .danger_accept_invalid_certs(true)
+            .danger_accept_invalid_hostnames(true);
+
+        builder.try_into().expect("Couldn't build a TLS connector!")
+    }
+}
+
 use crate::traits::*;
 use async_trait::async_trait;
 use futures::Future;
@@ -237,24 +260,6 @@ macro_rules! implement_traits_for {
             }
         }
 
-        impl TlsProvider<net::TcpStream> for $runtime {
-            type TlsStream = tls::TlsStream;
-            type Connector = tls::TlsConnector;
-
-            fn tls_connector(&self) -> tls::TlsConnector {
-                let mut builder = native_tls::TlsConnector::builder();
-                // These function names are scary, but they just mean that we
-                // aren't checking whether the signer of this cert
-                // participates in the web PKI, and we aren't checking the
-                // hostname in the cert.
-                builder
-                    .danger_accept_invalid_certs(true)
-                    .danger_accept_invalid_hostnames(true);
-
-                builder.try_into().expect("Couldn't build a TLS connector!")
-            }
-        }
-
         #[async_trait]
         impl crate::traits::TcpProvider for $runtime {
             type TcpStream = net::TcpStream;
@@ -273,7 +278,7 @@ macro_rules! implement_traits_for {
 }
 
 /// Create and return a new Tokio multithreaded runtime.
-pub fn create_runtime() -> IoResult<async_executors::TokioTp> {
+pub(crate) fn create_runtime() -> IoResult<async_executors::TokioTp> {
     let mut builder = async_executors::TokioTpBuilder::new();
     builder.tokio_builder().enable_all();
     builder.build()
@@ -281,11 +286,16 @@ pub fn create_runtime() -> IoResult<async_executors::TokioTp> {
 
 /// Wrapper around a Handle to a tokio runtime.
 ///
+/// Ideally, this type would go away, and we would just use
+/// `tokio::runtime::Handle` directly.  Unfortunately, we can't implement
+/// `futures::Spawn` on it ourselves because of Rust's orphan rules, so we need
+/// to define a new type here.
+///
 /// # Limitations
 ///
-/// Note that Arti requires that the runtime should have working
-/// implementations for Tokio's time, net, and io facilities, but we have
-/// no good way to check that when creating this object.
+/// Note that Arti requires that the runtime should have working implementations
+/// for Tokio's time, net, and io facilities, but we have no good way to check
+/// that when creating this object.
 #[derive(Clone, Debug)]
 pub struct TokioRuntimeHandle {
     /// The underlying Handle.
@@ -300,7 +310,7 @@ impl TokioRuntimeHandle {
     /// Note that Arti requires that the runtime should have working
     /// implementations for Tokio's time, net, and io facilities, but we have
     /// no good way to check that when creating this object.
-    pub fn new(handle: tokio_crate::runtime::Handle) -> Self {
+    pub(crate) fn new(handle: tokio_crate::runtime::Handle) -> Self {
         handle.into()
     }
 }

crates/tor-rtcompat/src/lib.rs

@@ -61,8 +61,8 @@
 //!     using for anything besides Arti, you can use [`create_runtime()`].
 //!
 //!   * If you want to explicitly construct a runtime with a specific
-//!     backend, you can do so with [`async_std::create_async_std_runtime`] or
-//!     [`tokio::create_tokio_runtime`].  Or if you have already constructed a
+//!     backend, you can do so with [`async_std::create_runtime`] or
+//!     [`tokio::create_runtime`].  Or if you have already constructed a
 //!     tokio runtime that you want to use, you can wrap it as a
 //!     [`Runtime`] explicitly with [`tokio::TokioRuntimeHandle`].
 //!
@@ -208,7 +208,7 @@ pub fn current_user_runtime() -> std::io::Result<impl Runtime> {
 ///
 /// Tokio users may want to avoid this function and instead make a
 /// runtime using [`current_user_runtime()`] or
-/// [`tokio::TokioRuntimeHandle::new()`]: this function always _builds_ a
+/// [`tokio::current_runtime()`]: this function always _builds_ a
 /// runtime, and if you already have a runtime, that isn't what you
 /// want with Tokio.
 ///

crates/tor-rtcompat/src/opaque.rs

@@ -55,6 +55,12 @@ macro_rules! implement_opaque_runtime {
         }
     }
 
+    impl std::fmt::Debug for $t {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            f.debug_struct(stringify!($t)).finish_non_exhaustive()
+        }
+    }
+
     // This boilerplate will fail unless $t implements Runtime.
     const _ : () = {
         fn assert_runtime<R: $crate::Runtime>() {}

crates/tor-rtcompat/src/tokio.rs

 //! Entry points for use with Tokio runtimes.
-pub use crate::impls::tokio::create_runtime as create_tokio_runtime;
-pub use crate::impls::tokio::TokioRuntimeHandle;
+use crate::impls::tokio::{NativeTlsTokio, TokioRuntimeHandle as Handle};
+use async_executors::TokioTp;
 
-use crate::Runtime;
-use std::io::{Error as IoError, ErrorKind};
+use crate::{CompoundRuntime, Runtime, SpawnBlocking};
+use std::io::{Error as IoError, ErrorKind, Result as IoResult};
+
+/// A [`Runtime`] built around a Handle to a tokio runtime, and `native_tls`.
+///
+/// # Limitations
+///
+/// Note that Arti requires that the runtime should have working
+/// implementations for Tokio's time, net, and io facilities, but we have
+/// no good way to check that when creating this object.
+#[derive(Clone)]
+pub struct TokioRuntimeHandle {
+    /// The actual [`CompoundRuntime`] that implements this.
+    inner: HandleInner,
+}
+
+/// Implementation type for a TokioRuntimeHandle.
+type HandleInner = CompoundRuntime<Handle, Handle, Handle, NativeTlsTokio>;
+
+/// A [`Runtime`] built around an owned `TokioTp` executor, and `native_tls`.
+#[derive(Clone)]
+pub struct TokioRuntime {
+    /// The actual [`CompoundRuntime`] that implements this.
+    inner: TokioRuntimeInner,
+}
+
+/// Implementation type for TokioRuntime.
+type TokioRuntimeInner = CompoundRuntime<TokioTp, TokioTp, TokioTp, NativeTlsTokio>;
+
+crate::opaque::implement_opaque_runtime! {
+    TokioRuntimeHandle { inner : HandleInner }
+}
+
+crate::opaque::implement_opaque_runtime! {
+    TokioRuntime { inner : TokioRuntimeInner }
+}
+
+impl From<tokio_crate::runtime::Handle> for TokioRuntimeHandle {
+    fn from(h: tokio_crate::runtime::Handle) -> Self {
+        let h = Handle::new(h);
+        TokioRuntimeHandle {
+            inner: CompoundRuntime::new(h.clone(), h.clone(), h, NativeTlsTokio::default()),
+        }
+    }
+}
+
+/// Create and return a new Tokio multithreaded runtime.
+fn create_tokio_runtime() -> IoResult<TokioRuntime> {
+    crate::impls::tokio::create_runtime().map(|r| TokioRuntime {
+        inner: CompoundRuntime::new(r.clone(), r.clone(), r, NativeTlsTokio::default()),
+    })
+}
 
 /// Create a new Tokio-based [`Runtime`].
 ///
@@ -12,8 +62,7 @@ use std::io::{Error as IoError, ErrorKind};
 /// runtime.
 ///
 /// Tokio users may want to avoid this function and instead make a
-/// runtime using [`current_runtime()`] or
-/// [`TokioRuntimeHandle::new()`]: this function always _builds_ a
+/// runtime using [`current_runtime()`]: this function always _builds_ a
 /// runtime, and if you already have a runtime, that isn't what you
 /// want with Tokio.
 pub fn create_runtime() -> std::io::Result<impl Runtime> {
@@ -35,7 +84,10 @@ pub fn create_runtime() -> std::io::Result<impl Runtime> {
 pub fn current_runtime() -> std::io::Result<TokioRuntimeHandle> {
     let handle = tokio_crate::runtime::Handle::try_current()
         .map_err(|e| IoError::new(ErrorKind::Other, e))?;
-    Ok(TokioRuntimeHandle::new(handle))
+    let h = Handle::new(handle);
+    Ok(TokioRuntimeHandle {
+        inner: CompoundRuntime::new(h.clone(), h.clone(), h, NativeTlsTokio::default()),
+    })
 }
 
 /// Run a test function using a freshly created tokio runtime.
@@ -45,9 +97,9 @@ pub fn current_runtime() -> std::io::Result<TokioRuntimeHandle> {
 /// Panics if we can't create a tokio runtime.
 pub fn test_with_runtime<P, F, O>(func: P) -> O
 where
-    P: FnOnce(async_executors::TokioTp) -> F,
+    P: FnOnce(TokioRuntime) -> F,
     F: futures::Future<Output = O>,
 {
     let runtime = create_tokio_runtime().expect("Failed to create a tokio runtime");
-    runtime.block_on(func(runtime.clone()))
+    runtime.clone().block_on(func(runtime))
 }