adding monitor

Author: win-t

src/rt/mod.rs

@@ -9,6 +9,7 @@ use crate::utils::abort_on_panic;
 pub use reactor::{Reactor, Watcher};
 pub use runtime::Runtime;
 
+mod monitor;
 mod reactor;
 mod runtime;
 
@@ -21,3 +22,11 @@ pub static RUNTIME: Lazy<Runtime> = Lazy::new(|| {
 
     Runtime::new()
 });
+
+pub fn scale_up() {
+    RUNTIME.scale_up();
+}
+
+pub fn scale_down() {
+    RUNTIME.scale_down();
+}

src/rt/monitor.rs

@@ -0,0 +1,42 @@
+//! Monitor for the runtime.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread;
+use std::time::{Duration, Instant};
+
+use crate::rt;
+use crate::task;
+
+pub fn run() {
+    const PROB_INTERVAL: Duration = Duration::from_millis(500);
+    const SCALE_DOWN_INTERVAL: Duration = Duration::from_secs(5);
+
+    let running = &Arc::new(AtomicBool::new(false));
+
+    {
+        let running = Arc::clone(running);
+        task::spawn(async move {
+            loop {
+                running.store(true, Ordering::SeqCst);
+                task::sleep(PROB_INTERVAL).await;
+            }
+        });
+    }
+
+    let mut next_scalling_down = Instant::now() + SCALE_DOWN_INTERVAL;
+
+    loop {
+        running.store(false, Ordering::SeqCst);
+        thread::sleep(PROB_INTERVAL + Duration::from_millis(10));
+        if !running.load(Ordering::SeqCst) {
+            eprintln!("WARNING: You are blocking the runtime, please use spawn_blocking");
+            rt::scale_up();
+        }
+
+        if next_scalling_down <= Instant::now() {
+            rt::scale_down();
+            next_scalling_down += SCALE_DOWN_INTERVAL;
+        }
+    }
+}

src/rt/runtime.rs

@@ -12,6 +12,7 @@ use crossbeam_deque::{Injector, Steal, Stealer, Worker};
 use crossbeam_utils::thread::{scope, Scope};
 use once_cell::unsync::OnceCell;
 
+use crate::rt::monitor;
 use crate::rt::Reactor;
 use crate::sync::Spinlock;
 use crate::task::Runnable;
@@ -40,7 +41,6 @@ enum Task {
 enum Action {
     ScaleUp,
     ScaleDown,
-    Terminate,
 }
 
 /// An async runtime.
@@ -138,6 +138,16 @@ impl Runtime {
         scope(|s| {
             (0..self.min_worker).for_each(|_| self.start_new_thread(s));
 
+            s.builder()
+                .name("async-std/monitor".to_string())
+                .spawn(move |_| {
+                    abort_on_panic(|| {
+                        monitor::run();
+                        panic!("Monitor function must not return");
+                    })
+                })
+                .expect("cannot start a monitor thread");
+
             loop {
                 match self.reciever.recv().unwrap() {
                     Action::ScaleUp => self.start_new_thread(s),
@@ -146,7 +156,6 @@ impl Runtime {
                         // and terminate itself
                         self.injector.push(Task::Terminate)
                     }
-                    Action::Terminate => return,
                 }
             }
         })
@@ -179,9 +188,6 @@ impl Runtime {
         }
         if let Some(index) = index {
             stealers.remove(index);
-            if stealers.is_empty() {
-                self.sender.send(Action::Terminate).unwrap();
-            }
         }
     }