RUST-1804 Replace async_once with tokio::sync::OnceCell (#992)

Author: Expyron

.evergreen/MSRV-Cargo.lock

@@ -181,7 +181,6 @@ features = ["v4"]
 [dev-dependencies]
 anyhow = { version = "1.0", features = ["backtrace"] }
 approx = "0.5.1"
-async_once = "0.2.6"
 backtrace = { version = "0.3.68" }
 ctrlc = "3.2.2"
 function_name = "0.2.1"
@@ -194,7 +193,7 @@ serde = { version = ">= 0.0.0", features = ["rc"] }
 serde_json = "1.0.64"
 semver = "1.0.0"
 time = "0.3.9"
-tokio = { version = ">= 0.0.0", features = ["fs"] }
+tokio = { version = ">= 0.0.0", features = ["fs", "parking_lot"] }
 tracing-subscriber = "0.3.16"
 regex = "1.6.0"
 serde-hex = "0.1.0"

Cargo.toml

@@ -14,13 +14,13 @@ use crate::{
     sdam::ServerInfo,
     selection_criteria::SelectionCriteria,
     test::{
+        get_client_options,
         log_uncaptured,
         Event,
         EventClient,
         EventHandler,
         SdamEvent,
         TestClient,
-        CLIENT_OPTIONS,
     },
     Client,
     Collection,
@@ -307,7 +307,7 @@ async fn cluster_time_in_commands() {
     }
 
     let handler = Arc::new(EventHandler::new());
-    let mut options = CLIENT_OPTIONS.get().await.clone();
+    let mut options = get_client_options().await.clone();
     options.heartbeat_freq = Some(Duration::from_secs(1000));
     options.command_event_handler = Some(handler.clone());
     options.sdam_event_handler = Some(handler.clone());

src/client/session/test.rs

@@ -27,12 +27,12 @@ use crate::{
     test::{
         assert_matches,
         eq_matches,
+        get_client_options,
         log_uncaptured,
         run_spec_test,
         EventClient,
         MatchErrExt,
         Matchable,
-        CLIENT_OPTIONS,
     },
 };
 use bson::doc;
@@ -157,9 +157,9 @@ impl Executor {
         let (updater, mut receiver) = TopologyUpdater::channel();
 
         let pool = ConnectionPool::new(
-            CLIENT_OPTIONS.get().await.hosts[0].clone(),
+            get_client_options().await.hosts[0].clone(),
             ConnectionEstablisher::new(EstablisherOptions::from_client_options(
-                CLIENT_OPTIONS.get().await,
+                get_client_options().await,
             ))
             .unwrap(),
             updater,
@@ -433,7 +433,7 @@ async fn cmap_spec_tests() {
             return;
         }
 
-        let mut options = CLIENT_OPTIONS.get().await.clone();
+        let mut options = get_client_options().await.clone();
         if options.load_balanced.unwrap_or(false) {
             log_uncaptured(format!(
                 "skipping {:?} due to load balanced topology",

src/cmap/test.rs

@@ -16,12 +16,12 @@ use crate::{
     sdam::TopologyUpdater,
     selection_criteria::ReadPreference,
     test::{
+        get_client_options,
         log_uncaptured,
         FailCommandOptions,
         FailPoint,
         FailPointMode,
         TestClient,
-        CLIENT_OPTIONS,
     },
 };
 use semver::VersionReq;
@@ -40,7 +40,7 @@ struct DatabaseEntry {
 #[cfg_attr(feature = "tokio-runtime", tokio::test)]
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 async fn acquire_connection_and_send_command() {
-    let client_options = CLIENT_OPTIONS.get().await.clone();
+    let client_options = get_client_options().await.clone();
     let mut pool_options = ConnectionPoolOptions::from_client_options(&client_options);
     pool_options.ready = Some(true);
 
@@ -84,7 +84,7 @@ async fn acquire_connection_and_send_command() {
 #[cfg_attr(feature = "tokio-runtime", tokio::test)]
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 async fn concurrent_connections() {
-    let mut options = CLIENT_OPTIONS.get().await.clone();
+    let mut options = get_client_options().await.clone();
     if options.load_balanced.unwrap_or(false) {
         log_uncaptured("skipping concurrent_connections test due to load-balanced topology");
         return;
@@ -115,14 +115,14 @@ async fn concurrent_connections() {
         .expect("failpoint should succeed");
 
     let handler = Arc::new(EventHandler::new());
-    let client_options = CLIENT_OPTIONS.get().await.clone();
+    let client_options = get_client_options().await.clone();
     let mut options = ConnectionPoolOptions::from_client_options(&client_options);
     options.cmap_event_handler =
         Some(handler.clone() as Arc<dyn crate::event::cmap::CmapEventHandler>);
     options.ready = Some(true);
 
     let pool = ConnectionPool::new(
-        CLIENT_OPTIONS.get().await.hosts[0].clone(),
+        get_client_options().await.hosts[0].clone(),
         ConnectionEstablisher::new(EstablisherOptions::from_client_options(&client_options))
             .unwrap(),
         TopologyUpdater::channel().0,
@@ -174,7 +174,7 @@ async fn concurrent_connections() {
 #[function_name::named]
 
 async fn connection_error_during_establishment() {
-    let mut client_options = CLIENT_OPTIONS.get().await.clone();
+    let mut client_options = get_client_options().await.clone();
     if client_options.load_balanced.unwrap_or(false) {
         log_uncaptured(
             "skipping connection_error_during_establishment test due to load-balanced topology",
@@ -237,7 +237,7 @@ async fn connection_error_during_establishment() {
 #[function_name::named]
 
 async fn connection_error_during_operation() {
-    let mut options = CLIENT_OPTIONS.get().await.clone();
+    let mut options = get_client_options().await.clone();
     let handler = Arc::new(EventHandler::new());
     options.cmap_event_handler = Some(handler.clone() as Arc<dyn CmapEventHandler>);
     options.hosts.drain(1..);

src/cmap/test/integration.rs

@@ -9,7 +9,7 @@ use bson::{doc, Bson};
 use crate::{
     client::options::ClientOptions,
     compression::{Compressor, CompressorId, Decoder},
-    test::{TestClient, CLIENT_OPTIONS},
+    test::{get_client_options, TestClient},
 };
 
 #[cfg(feature = "zlib-compression")]
@@ -67,7 +67,7 @@ fn test_snappy_compressor() {
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[cfg(feature = "zlib-compression")]
 async fn ping_server_with_zlib_compression() {
-    let mut client_options = CLIENT_OPTIONS.get().await.clone();
+    let mut client_options = get_client_options().await.clone();
     client_options.compressors = Some(vec![Compressor::Zlib { level: Some(4) }]);
     send_ping_with_compression(client_options).await;
 }
@@ -76,7 +76,7 @@ async fn ping_server_with_zlib_compression() {
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[cfg(feature = "zstd-compression")]
 async fn ping_server_with_zstd_compression() {
-    let mut client_options = CLIENT_OPTIONS.get().await.clone();
+    let mut client_options = get_client_options().await.clone();
     client_options.compressors = Some(vec![Compressor::Zstd { level: None }]);
     send_ping_with_compression(client_options).await;
 }
@@ -85,7 +85,7 @@ async fn ping_server_with_zstd_compression() {
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 #[cfg(feature = "snappy-compression")]
 async fn ping_server_with_snappy_compression() {
-    let mut client_options = CLIENT_OPTIONS.get().await.clone();
+    let mut client_options = get_client_options().await.clone();
     client_options.compressors = Some(vec![Compressor::Snappy]);
     send_ping_with_compression(client_options).await;
 }
@@ -98,7 +98,7 @@ async fn ping_server_with_snappy_compression() {
     feature = "snappy-compression"
 ))]
 async fn ping_server_with_all_compressors() {
-    let mut client_options = CLIENT_OPTIONS.get().await.clone();
+    let mut client_options = get_client_options().await.clone();
     client_options.compressors = Some(vec![
         Compressor::Zlib { level: None },
         Compressor::Snappy,

src/compression/test.rs

@@ -16,6 +16,7 @@ use crate::{
     sdam::{description::topology::server_selection, Server},
     selection_criteria::{ReadPreference, SelectionCriteria},
     test::{
+        get_client_options,
         log_uncaptured,
         run_spec_test,
         Event,
@@ -24,7 +25,6 @@ use crate::{
         FailPoint,
         FailPointMode,
         TestClient,
-        CLIENT_OPTIONS,
     },
     ServerInfo,
 };
@@ -119,7 +119,7 @@ async fn select_in_window() {
 #[cfg_attr(feature = "tokio-runtime", tokio::test(flavor = "multi_thread"))]
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 async fn load_balancing_test() {
-    let mut setup_client_options = CLIENT_OPTIONS.get().await.clone();
+    let mut setup_client_options = get_client_options().await.clone();
 
     if setup_client_options.load_balanced.unwrap_or(false) {
         log_uncaptured("skipping load_balancing_test test due to load-balanced topology");
@@ -149,7 +149,7 @@ async fn load_balancing_test() {
         return;
     }
 
-    if CLIENT_OPTIONS.get().await.hosts.len() != 2 {
+    if get_client_options().await.hosts.len() != 2 {
         log_uncaptured("skipping load_balancing_test test due to topology not having 2 mongoses");
         return;
     }
@@ -216,7 +216,7 @@ async fn load_balancing_test() {
 
     let mut handler = EventHandler::new();
     let mut subscriber = handler.subscribe();
-    let mut options = CLIENT_OPTIONS.get().await.clone();
+    let mut options = get_client_options().await.clone();
     let max_pool_size = DEFAULT_MAX_POOL_SIZE;
     let hosts = options.hosts.clone();
     options.local_threshold = Duration::from_secs(30).into();
@@ -260,7 +260,7 @@ async fn load_balancing_test() {
         .build();
     let failpoint = FailPoint::fail_command(&["find"], FailPointMode::AlwaysOn, options);
 
-    let slow_host = CLIENT_OPTIONS.get().await.hosts[0].clone();
+    let slow_host = get_client_options().await.hosts[0].clone();
     let criteria = SelectionCriteria::Predicate(Arc::new(move |si| si.address() == &slow_host));
     let fp_guard = setup_client
         .enable_failpoint(failpoint, criteria)

src/sdam/description/topology/server_selection/test/in_window.rs

@@ -24,6 +24,7 @@ use crate::{
     },
     selection_criteria::TagSet,
     test::{
+        get_client_options,
         log_uncaptured,
         run_spec_test,
         Event,
@@ -34,7 +35,6 @@ use crate::{
         FailPointMode,
         SdamEvent,
         TestClient,
-        CLIENT_OPTIONS,
     },
 };
 
@@ -638,7 +638,7 @@ async fn topology_closed_event_last() {
 #[cfg_attr(feature = "tokio-runtime", tokio::test(flavor = "multi_thread"))]
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 async fn heartbeat_events() {
-    let mut options = CLIENT_OPTIONS.get().await.clone();
+    let mut options = get_client_options().await.clone();
     options.hosts.drain(1..);
     options.heartbeat_freq = Some(Duration::from_millis(50));
     options.app_name = "heartbeat_events".to_string().into();
@@ -721,7 +721,7 @@ async fn direct_connection() {
         .await
         .expect("failed to select secondary");
 
-    let mut secondary_options = CLIENT_OPTIONS.get().await.clone();
+    let mut secondary_options = get_client_options().await.clone();
     secondary_options.hosts = vec![secondary_address];
 
     let mut direct_false_options = secondary_options.clone();

src/sdam/description/topology/test/sdam.rs

@@ -8,7 +8,7 @@ use crate::{
     options::{ClientOptions, ServerAddress},
     runtime,
     sdam::Topology,
-    test::{log_uncaptured, CLIENT_OPTIONS},
+    test::{get_client_options, log_uncaptured},
 };
 
 fn localhost_test_build_10gen(port: u16) -> ServerAddress {
@@ -131,7 +131,7 @@ async fn no_results() {
 #[cfg_attr(feature = "tokio-runtime", tokio::test)]
 #[cfg_attr(feature = "async-std-runtime", async_std::test)]
 async fn load_balanced_no_srv_polling() {
-    if CLIENT_OPTIONS.get().await.load_balanced != Some(true) {
+    if get_client_options().await.load_balanced != Some(true) {
         log_uncaptured("skipping load_balanced_no_srv_polling due to not load balanced topology");
         return;
     }