RUST-995 Use the tokio runtime for the sync API

Author: isabelatkinson

Cargo.toml

@@ -23,7 +23,7 @@ exclude = [
 default = ["tokio-runtime"]
 tokio-runtime = ["tokio/macros", "tokio/net", "tokio/rt", "tokio/time", "serde_bytes"]
 async-std-runtime = ["async-std", "async-std/attributes", "async-std-resolver", "tokio-util/compat"]
-sync = ["async-std-runtime"]
+sync = ["tokio-runtime"]
 
 # Enable support for v0.4 of the chrono crate in the public API of the BSON library.
 bson-chrono-0_4 = ["bson/chrono-0_4"]
@@ -47,6 +47,7 @@ bson = { git = "https://github.com/mongodb/bson-rust", branch = "master" }
 chrono = "0.4.7"
 derivative = "2.1.1"
 flate2 = { version = "1.0", optional = true }
+futures = "0.3"
 futures-core = "0.3.14"
 futures-io = "0.3.14"
 futures-util = { version = "0.3.14", features = ["io"] }
@@ -126,7 +127,6 @@ features = ["v4"]
 approx = "0.4.0"
 derive_more = "0.99.13"
 function_name = "0.2.0"
-futures = "0.3"
 home = "0.5"
 pretty_assertions = "0.7.1"
 serde_json = "1.0.64"

src/client/options/mod.rs

@@ -1048,7 +1048,7 @@ impl ClientOptions {
     #[cfg(any(feature = "sync", docsrs))]
     #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
     pub fn parse(s: impl AsRef<str>) -> Result<Self> {
-        crate::RUNTIME.block_on(Self::parse_uri(s.as_ref(), None))
+        crate::sync::TOKIO_RUNTIME.block_on(Self::parse_uri(s.as_ref(), None))
     }
 
     /// Parses a MongoDB connection string into a `ClientOptions` struct.
@@ -1078,7 +1078,7 @@ impl ClientOptions {
     #[cfg(any(feature = "sync", docsrs))]
     #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
     pub fn parse_with_resolver_config(uri: &str, resolver_config: ResolverConfig) -> Result<Self> {
-        crate::RUNTIME.block_on(Self::parse_uri(uri, Some(resolver_config)))
+        crate::sync::TOKIO_RUNTIME.block_on(Self::parse_uri(uri, Some(resolver_config)))
     }
 
     /// Populate this `ClientOptions` from the given URI, optionally using the resolver config for

src/cmap/establish/handshake/mod.rs

@@ -18,14 +18,14 @@ use crate::{
     sdam::Topology,
 };
 
-#[cfg(feature = "tokio-runtime")]
+#[cfg(all(feature = "tokio-runtime", not(feature = "sync")))]
 const RUNTIME_NAME: &str = "tokio";
 
-#[cfg(all(feature = "async-std-runtime", not(feature = "sync")))]
+#[cfg(feature = "async-std-runtime")]
 const RUNTIME_NAME: &str = "async-std";
 
 #[cfg(feature = "sync")]
-const RUNTIME_NAME: &str = "sync (with async-std)";
+const RUNTIME_NAME: &str = "sync (with tokio)";
 
 #[derive(Clone, Debug)]
 struct ClientMetadata {

src/lib.rs

@@ -367,10 +367,9 @@ compile_error!(
      `async-std-runtime` or set `default-features = false` in your Cargo.toml"
 );
 
-#[cfg(all(feature = "tokio-runtime", feature = "sync"))]
+#[cfg(all(feature = "async-std-runtime", feature = "sync"))]
 compile_error!(
-    "`tokio-runtime` and `sync` can't both be enabled; either disable `sync` or set \
-     `default-features = false` in your Cargo.toml"
+    "`async-std-runtime` and `sync` can't both be enabled"
 );
 
 #[cfg(all(not(feature = "tokio-runtime"), not(feature = "async-std-runtime")))]

src/runtime/mod.rs

@@ -51,12 +51,15 @@ impl AsyncRuntime {
     {
         match self {
             #[cfg(feature = "tokio-runtime")]
-            Self::Tokio => match TokioCallingContext::current() {
-                TokioCallingContext::Async(handle) => {
-                    Some(AsyncJoinHandle::Tokio(handle.spawn(fut)))
-                }
-                TokioCallingContext::Sync => None,
-            },
+            Self::Tokio => {
+                #[cfg(feature = "sync")]
+                let handle = crate::sync::TOKIO_RUNTIME.handle();
+
+                #[cfg(not(feature = "sync"))]
+                let handle = tokio::runtime::Handle::current();
+
+                Some(AsyncJoinHandle::Tokio(handle.spawn(fut)))
+            }
 
             #[cfg(feature = "async-std-runtime")]
             Self::AsyncStd => Some(AsyncJoinHandle::AsyncStd(async_std::task::spawn(fut))),
@@ -74,32 +77,6 @@ impl AsyncRuntime {
         self.spawn(fut);
     }
 
-    /// Run a future in the foreground, blocking on it completing.
-    ///
-    /// This will panic if called from a sychronous context when tokio is being used.
-    #[cfg(any(feature = "sync", test))]
-    pub(crate) fn block_on<F, T>(self, fut: F) -> T
-    where
-        F: Future<Output = T>,
-    {
-        #[cfg(all(feature = "tokio-runtime", not(feature = "async-std-runtime")))]
-        {
-            match TokioCallingContext::current() {
-                TokioCallingContext::Async(_handle) => {
-                    tokio::task::block_in_place(|| futures::executor::block_on(fut))
-                }
-                TokioCallingContext::Sync => {
-                    panic!("block_on called from tokio outside of async context")
-                }
-            }
-        }
-
-        #[cfg(feature = "async-std-runtime")]
-        {
-            async_std::task::block_on(fut)
-        }
-    }
-
     /// Run a future in the foreground, blocking on it completing.
     /// This does not notify the runtime that it will be blocking and should only be used for
     /// operations that will immediately (or quickly) succeed.
@@ -177,25 +154,3 @@ impl AsyncRuntime {
         }
     }
 }
-
-/// Represents the context in which a given runtime method is being called from.
-#[cfg(feature = "tokio-runtime")]
-enum TokioCallingContext {
-    /// From a syncronous setting (i.e. not from a runtime thread).
-    Sync,
-
-    /// From an asyncronous setting (i.e. from an async block or function being run on a runtime).
-    /// Includes a handle to the current runtime.
-    Async(tokio::runtime::Handle),
-}
-
-#[cfg(feature = "tokio-runtime")]
-impl TokioCallingContext {
-    /// Get the current calling context.
-    fn current() -> Self {
-        match tokio::runtime::Handle::try_current() {
-            Ok(handle) => TokioCallingContext::Async(handle),
-            Err(_) => TokioCallingContext::Sync,
-        }
-    }
-}

src/sync/change_stream.rs

@@ -8,7 +8,7 @@ use crate::{
         ChangeStream as AsyncChangeStream,
     },
     error::Result,
-    RUNTIME,
+    sync::TOKIO_RUNTIME,
 };
 
 use super::ClientSession;
@@ -112,7 +112,7 @@ where
     /// # }
     /// ```
     pub fn next_if_any(&mut self) -> Result<Option<T>> {
-        RUNTIME.block_on(self.async_stream.next_if_any())
+        TOKIO_RUNTIME.block_on(self.async_stream.next_if_any())
     }
 }
 
@@ -123,7 +123,7 @@ where
     type Item = Result<T>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        RUNTIME.block_on(self.async_stream.next())
+        TOKIO_RUNTIME.block_on(self.async_stream.next())
     }
 }
 
@@ -200,7 +200,7 @@ where
     /// # }
     /// ```
     pub fn next(&mut self, session: &mut ClientSession) -> Result<Option<T>> {
-        RUNTIME.block_on(self.async_stream.next(&mut session.async_client_session))
+        TOKIO_RUNTIME.block_on(self.async_stream.next(&mut session.async_client_session))
     }
 
     /// Returns whether the change stream will continue to receive events.
@@ -234,7 +234,7 @@ where
     /// # }
     /// ```
     pub fn next_if_any(&mut self, session: &mut ClientSession) -> Result<Option<T>> {
-        RUNTIME.block_on(
+        TOKIO_RUNTIME.block_on(
             self.async_stream
                 .next_if_any(&mut session.async_client_session),
         )

src/sync/client/mod.rs

@@ -14,8 +14,8 @@ use crate::{
         SessionOptions,
     },
     results::DatabaseSpecification,
+    sync::TOKIO_RUNTIME,
     Client as AsyncClient,
-    RUNTIME,
 };
 
 /// This is the main entry point for the synchronous API. A `Client` is used to connect to a MongoDB
@@ -75,7 +75,7 @@ impl Client {
     /// [`ClientOptions::parse`](../options/struct.ClientOptions.html#method.parse) for more
     /// details.
     pub fn with_uri_str(uri: impl AsRef<str>) -> Result<Self> {
-        let async_client = RUNTIME.block_on(AsyncClient::with_uri_str(uri.as_ref()))?;
+        let async_client = TOKIO_RUNTIME.block_on(AsyncClient::with_uri_str(uri.as_ref()))?;
         Ok(Self { async_client })
     }
 
@@ -134,7 +134,7 @@ impl Client {
         filter: impl Into<Option<Document>>,
         options: impl Into<Option<ListDatabasesOptions>>,
     ) -> Result<Vec<DatabaseSpecification>> {
-        RUNTIME.block_on(
+        TOKIO_RUNTIME.block_on(
             self.async_client
                 .list_databases(filter.into(), options.into()),
         )
@@ -146,15 +146,15 @@ impl Client {
         filter: impl Into<Option<Document>>,
         options: impl Into<Option<ListDatabasesOptions>>,
     ) -> Result<Vec<String>> {
-        RUNTIME.block_on(
+        TOKIO_RUNTIME.block_on(
             self.async_client
                 .list_database_names(filter.into(), options.into()),
         )
     }
 
     /// Starts a new `ClientSession`.
     pub fn start_session(&self, options: Option<SessionOptions>) -> Result<ClientSession> {
-        RUNTIME
+        TOKIO_RUNTIME
             .block_on(self.async_client.start_session(options))
             .map(Into::into)
     }
@@ -182,7 +182,7 @@ impl Client {
         pipeline: impl IntoIterator<Item = Document>,
         options: impl Into<Option<ChangeStreamOptions>>,
     ) -> Result<ChangeStream<ChangeStreamEvent<Document>>> {
-        RUNTIME
+        TOKIO_RUNTIME
             .block_on(self.async_client.watch(pipeline, options))
             .map(ChangeStream::new)
     }
@@ -195,7 +195,7 @@ impl Client {
         options: impl Into<Option<ChangeStreamOptions>>,
         session: &mut ClientSession,
     ) -> Result<SessionChangeStream<ChangeStreamEvent<Document>>> {
-        RUNTIME
+        TOKIO_RUNTIME
             .block_on(self.async_client.watch_with_session(
                 pipeline,
                 options,

src/sync/client/session.rs

@@ -4,8 +4,8 @@ use crate::{
     client::session::ClusterTime,
     error::Result,
     options::{SessionOptions, TransactionOptions},
+    sync::TOKIO_RUNTIME,
     ClientSession as AsyncClientSession,
-    RUNTIME,
 };
 
 /// A MongoDB client session. This struct represents a logical session used for ordering sequential
@@ -76,7 +76,7 @@ impl ClientSession {
         &mut self,
         options: impl Into<Option<TransactionOptions>>,
     ) -> Result<()> {
-        RUNTIME.block_on(self.async_client_session.start_transaction(options))
+        TOKIO_RUNTIME.block_on(self.async_client_session.start_transaction(options))
     }
 
     /// Commits the transaction that is currently active on this session.
@@ -100,7 +100,7 @@ impl ClientSession {
     /// [here](https://docs.mongodb.com/manual/core/retryable-writes/) for more information on
     /// retryable writes.
     pub fn commit_transaction(&mut self) -> Result<()> {
-        RUNTIME.block_on(self.async_client_session.commit_transaction())
+        TOKIO_RUNTIME.block_on(self.async_client_session.commit_transaction())
     }
 
     /// Aborts the transaction that is currently active on this session. Any open transaction will
@@ -133,6 +133,6 @@ impl ClientSession {
     /// [here](https://docs.mongodb.com/manual/core/retryable-writes/) for more information on
     /// retryable writes.
     pub fn abort_transaction(&mut self) -> Result<()> {
-        RUNTIME.block_on(self.async_client_session.abort_transaction())
+        TOKIO_RUNTIME.block_on(self.async_client_session.abort_transaction())
     }
 }