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;
 
@@ -19,5 +20,7 @@ pub static RUNTIME: Lazy<Runtime> = Lazy::new(|| {
         .spawn(|| abort_on_panic(|| RUNTIME.run()))
         .expect("cannot start a runtime thread");
 
+    monitor::spawn_thread();
+
     Runtime::new()
 });

src/rt/monitor.rs

@@ -0,0 +1,62 @@
+use std::cell::Cell;
+use std::future::Future;
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::Duration;
+
+use crate::rt::RUNTIME;
+use crate::task;
+use crate::utils::abort_on_panic;
+
+fn spawn_monitor_task<F>(future: F)
+where
+    F: Future<Output = ()> + Send + 'static,
+{
+    task::Builder::new()
+        .spawn_in_runtime(future, &RUNTIME)
+        .expect("cannot spawn monitor task");
+}
+
+pub fn spawn_thread() {
+    thread::Builder::new()
+        .name("async-std/monitor".to_string())
+        .spawn(|| {
+            const PROBING_DURATION_MS: u64 = 500;
+            const SCALING_DOWN_SEC: u64 = 1 * 60; // 1 minute
+
+            abort_on_panic(|| {
+                let running = &Arc::new(Mutex::new(Cell::new(false)));
+
+                {
+                    let running = Arc::clone(running);
+                    spawn_monitor_task(async move {
+                        loop {
+                            running.lock().unwrap().set(true);
+                            task::sleep(Duration::from_millis(PROBING_DURATION_MS)).await;
+                        }
+                    });
+                }
+
+                {
+                    spawn_monitor_task(async {
+                        loop {
+                            task::sleep(Duration::from_secs(SCALING_DOWN_SEC)).await;
+                            RUNTIME.scale_down();
+                        }
+                    });
+                }
+
+                loop {
+                    running.lock().unwrap().set(false);
+                    thread::sleep(Duration::from_millis(PROBING_DURATION_MS));
+                    if !running.lock().unwrap().get() {
+                        eprintln!(
+                            "WARNING: You are blocking the runtime, please use spawn_blocking"
+                        );
+                        RUNTIME.scale_up();
+                    }
+                }
+            })
+        })
+        .expect("cannot start a monitor thread");
+}

src/task/builder.rs

@@ -3,6 +3,7 @@ use std::future::Future;
 use kv_log_macro::trace;
 
 use crate::io;
+use crate::rt::Runtime;
 use crate::rt::RUNTIME;
 use crate::task::{JoinHandle, Task};
 use crate::utils::abort_on_panic;
@@ -29,6 +30,18 @@ impl Builder {
 
     /// Spawns a task with the configured settings.
     pub fn spawn<F, T>(self, future: F) -> io::Result<JoinHandle<T>>
+    where
+        F: Future<Output = T> + Send + 'static,
+        T: Send + 'static,
+    {
+        self.spawn_in_runtime(future, &RUNTIME)
+    }
+
+    pub(crate) fn spawn_in_runtime<F, T>(
+        self,
+        future: F,
+        rt: &'static Runtime,
+    ) -> io::Result<JoinHandle<T>>
     where
         F: Future<Output = T> + Send + 'static,
         T: Send + 'static,
@@ -58,7 +71,7 @@ impl Builder {
             future.await
         };
 
-        let schedule = move |t| RUNTIME.schedule(Runnable(t));
+        let schedule = move |t| rt.schedule(Runnable(t));
         let (task, handle) = async_task::spawn(future, schedule, task);
         task.schedule();
         Ok(JoinHandle::new(handle))