AsyncConcurrentDropper: use dedicated pool

It can deadlock easily if the drop itself schedules things in the same
pool (for e.g. dropping a sub component) and waits for it.

Author: msmouse

crates/aptos-drop-helper/src/async_concurrent_dropper.rs

@@ -2,10 +2,10 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::metrics::TIMER;
-use aptos_experimental_runtimes::thread_manager::THREAD_MANAGER;
 use aptos_infallible::Mutex;
 use aptos_metrics_core::TimerHelper;
 use std::sync::mpsc::{channel, Receiver, Sender};
+use threadpool::ThreadPool;
 
 /// A helper to send things to a thread pool for asynchronous dropping.
 ///
@@ -17,20 +17,24 @@ pub struct AsyncConcurrentDropper {
     name: &'static str,
     token_tx: Sender<()>,
     token_rx: Mutex<Receiver<()>>,
+    /// use dedicated threadpool to minimize the possibility of dead lock
+    thread_pool: ThreadPool,
 }
 
 impl AsyncConcurrentDropper {
-    pub fn new(name: &'static str, max_concurrent_drops: usize) -> Self {
+    pub fn new(name: &'static str, max_async_drops: usize, num_threads: usize) -> Self {
         let (token_tx, token_rx) = channel();
-        for _ in 0..max_concurrent_drops {
+        for _ in 0..max_async_drops {
             token_tx
                 .send(())
                 .expect("DropHelper: Failed to buffer initial tokens.");
         }
+        let thread_pool = ThreadPool::new(num_threads);
         Self {
             name,
             token_tx,
             token_rx: Mutex::new(token_rx),
+            thread_pool,
         }
     }
 
@@ -51,7 +55,7 @@ impl AsyncConcurrentDropper {
 
         let token_tx = self.token_tx.clone();
         let name = self.name;
-        THREAD_MANAGER.get_non_exe_cpu_pool().spawn(move || {
+        self.thread_pool.execute(move || {
             let _timer = TIMER.timer_with(&[name, "real_drop"]);
 
             drop(v);
@@ -74,32 +78,43 @@ mod tests {
 
     impl Drop for SlowDropper {
         fn drop(&mut self) {
-            sleep(Duration::from_secs(1));
+            sleep(Duration::from_millis(200));
         }
     }
 
     #[test]
-    fn test_concurrency_limit_hit() {
-        let s = AsyncConcurrentDropper::new("test", 8);
+    fn test_within_concurrency_limit() {
+        let s = AsyncConcurrentDropper::new("test", 8, 4);
         let now = std::time::Instant::now();
-        let rx = s.schedule_drop_with_waiter(SlowDropper);
-        for _ in 1..8 {
+        let rx1 = s.schedule_drop_with_waiter(SlowDropper); // first round
+        for _ in 0..3 {
+            s.schedule_drop(SlowDropper);
+        }
+        let rx2 = s.schedule_drop_with_waiter(SlowDropper); // second round
+        for _ in 0..3 {
             s.schedule_drop(SlowDropper);
         }
-        assert!(now.elapsed() < Duration::from_millis(500));
-        rx.recv().unwrap();
-        assert!(now.elapsed() > Duration::from_secs(1));
+        assert!(now.elapsed() < Duration::from_millis(200));
+        rx1.recv().unwrap();
+        assert!(now.elapsed() > Duration::from_millis(200));
+        assert!(now.elapsed() < Duration::from_millis(400));
+        rx2.recv().unwrap();
+        assert!(now.elapsed() > Duration::from_millis(400));
+        assert!(now.elapsed() < Duration::from_millis(600));
     }
 
     #[test]
-    fn test_within_concurrency_limit() {
-        let s = AsyncConcurrentDropper::new("test", 8);
+    fn test_concurrency_limit_hit() {
+        let s = AsyncConcurrentDropper::new("test", 8, 4);
         let now = std::time::Instant::now();
         for _ in 0..8 {
             s.schedule_drop(SlowDropper);
         }
-        assert!(now.elapsed() < Duration::from_millis(500));
+        assert!(now.elapsed() < Duration::from_millis(200));
+        s.schedule_drop(SlowDropper);
+        assert!(now.elapsed() > Duration::from_millis(200));
+        assert!(now.elapsed() < Duration::from_millis(400));
         s.schedule_drop(SlowDropper);
-        assert!(now.elapsed() > Duration::from_secs(1));
+        assert!(now.elapsed() < Duration::from_millis(400));
     }
 }

crates/aptos-drop-helper/src/lib.rs

@@ -9,4 +9,4 @@ pub mod async_drop_queue;
 mod metrics;
 
 pub static DEFAULT_DROPPER: Lazy<AsyncConcurrentDropper> =
-    Lazy::new(|| AsyncConcurrentDropper::new("default", 32));
+    Lazy::new(|| AsyncConcurrentDropper::new("default", 32, 8));