Replace once_cell with async-lock (#30)

Author: notgull

Cargo.toml

@@ -16,8 +16,8 @@ exclude = ["/.*"]
 
 [dependencies]
 async-channel = "1.4.0"
+async-lock = "2.6.0"
 async-task = "4.0.2"
 atomic-waker = "1.0.0"
 fastrand = "1.3.4"
 futures-lite = "1.11.0"
-once_cell = "1.4.1"

src/lib.rs

@@ -93,10 +93,10 @@ use std::thread;
 use std::time::Duration;
 
 use async_channel::{bounded, Receiver};
+use async_lock::OnceCell;
 use async_task::Runnable;
 use atomic_waker::AtomicWaker;
 use futures_lite::{future, prelude::*, ready};
-use once_cell::sync::Lazy;
 
 #[doc(no_inline)]
 pub use async_task::Task;
@@ -113,17 +113,6 @@ const MAX_MAX_THREADS: usize = 10000;
 /// Env variable that allows to override default value for max threads.
 const MAX_THREADS_ENV: &str = "BLOCKING_MAX_THREADS";
 
-/// Lazily initialized global executor.
-static EXECUTOR: Lazy<Executor> = Lazy::new(|| Executor {
-    inner: Mutex::new(Inner {
-        idle_count: 0,
-        thread_count: 0,
-        queue: VecDeque::new(),
-    }),
-    cvar: Condvar::new(),
-    thread_limit: Executor::max_threads(),
-});
-
 /// The blocking executor.
 struct Executor {
     /// Inner state of the executor.
@@ -162,11 +151,31 @@ impl Executor {
             Err(_) => DEFAULT_MAX_THREADS,
         }
     }
+
     /// Spawns a future onto this executor.
     ///
     /// Returns a [`Task`] handle for the spawned task.
     fn spawn<T: Send + 'static>(future: impl Future<Output = T> + Send + 'static) -> Task<T> {
-        let (runnable, task) = async_task::spawn(future, |r| EXECUTOR.schedule(r));
+        static EXECUTOR: OnceCell<Executor> = OnceCell::new();
+
+        let (runnable, task) = async_task::spawn(future, |r| {
+            // Initialize the executor if we haven't already.
+            let executor = EXECUTOR.get_or_init_blocking(|| {
+                let thread_limit = Self::max_threads();
+                Executor {
+                    inner: Mutex::new(Inner {
+                        idle_count: 0,
+                        thread_count: 0,
+                        queue: VecDeque::new(),
+                    }),
+                    cvar: Condvar::new(),
+                    thread_limit,
+                }
+            });
+
+            // Schedule the task on our executor.
+            executor.schedule(r)
+        });
         runnable.schedule();
         task
     }
@@ -223,8 +232,7 @@ impl Executor {
     fn grow_pool(&'static self, mut inner: MutexGuard<'static, Inner>) {
         // If runnable tasks greatly outnumber idle threads and there aren't too many threads
         // already, then be aggressive: wake all idle threads and spawn one more thread.
-        while inner.queue.len() > inner.idle_count * 5 && inner.thread_count < EXECUTOR.thread_limit
-        {
+        while inner.queue.len() > inner.idle_count * 5 && inner.thread_count < self.thread_limit {
             // The new thread starts in idle state.
             inner.idle_count += 1;
             inner.thread_count += 1;