feat(serverless): add LRU time cache for isolates

.changeset/few-foxes-cheat.md

@@ -0,0 +1,5 @@
+---
+'@lagon/serverless': patch
+---
+
+Add configurable LRU time cache for isolates

.changeset/tender-walls-tap.md

@@ -0,0 +1,5 @@
+---
+'@lagon/runtime': patch
+---
+
+Add on_drop callback and properly handle isolate termination

packages/runtime/src/isolate/mod.rs

@@ -94,13 +94,15 @@ pub struct IsolateStatistics {
     pub memory_usage: usize,
 }
 
-#[derive(Debug)]
+// #[derive(Debug)]
 pub struct IsolateOptions {
     pub code: String,
     pub environment_variables: Option<HashMap<String, String>>,
-    pub memory: usize, // in MB (MegaBytes)
+    pub memory: usize,  // in MB (MegaBytes)
     pub timeout: usize, // in ms (MilliSeconds)
-                       // pub snapshot_blob: Option<Box<dyn Allocated<[u8]>>>,
+    pub id: Option<String>,
+    pub on_drop: Option<Box<dyn Fn(Option<String>)>>,
+    // pub snapshot_blob: Option<Box<dyn Allocated<[u8]>>>,
 }
 
 impl IsolateOptions {
@@ -110,6 +112,8 @@ impl IsolateOptions {
             environment_variables: None,
             timeout: 50,
             memory: 128,
+            id: None,
+            on_drop: None,
             // snapshot_blob: None,
         }
     }
@@ -131,6 +135,16 @@ impl IsolateOptions {
         self.memory = memory;
         self
     }
+
+    pub fn with_id(mut self, id: String) -> Self {
+        self.id = Some(id);
+        self
+    }
+
+    pub fn with_on_drop_callback(mut self, on_drop: Box<dyn Fn(Option<String>)>) -> Self {
+        self.on_drop = Some(on_drop);
+        self
+    }
 }
 
 #[derive(Debug)]
@@ -549,6 +563,18 @@ impl Isolate {
     }
 }
 
+impl Drop for Isolate {
+    fn drop(&mut self) {
+        if !self.isolate.is_execution_terminating() {
+            self.isolate.terminate_execution();
+        }
+
+        if let Some(on_drop) = &self.options.on_drop {
+            on_drop(self.options.id.clone());
+        }
+    }
+}
+
 fn get_exception_message(
     scope: &mut v8::TryCatch<v8::HandleScope>,
     exception: v8::Local<v8::Value>,

packages/serverless/.env.example

@@ -1,8 +1,7 @@
 LAGON_TOKEN=
 LAGON_ROOT_DOMAIN=lagon.app
 LAGON_REGION=
-SENTRY_DSN=
-SENTRY_ENVIRONMENT=development
+LAGON_ISOLATES_CACHE_SECONDS=60
 
 DATABASE_URL=mysql://root:mysql@localhost:3306/lagon
 REDIS_URL=redis://localhost:6379

packages/serverless/Cargo.toml

@@ -21,3 +21,4 @@ axiom-rs = "0.6.0"
 chrono = "0.4.22"
 lazy_static = "1.4.0"
 rand = { version = "0.8.5", features = ["std_rng"] }
+lru_time_cache = "0.11.11"

packages/serverless/src/main.rs

@@ -9,7 +9,8 @@ use lagon_runtime::http::{Request, RunResult, StreamResult};
 use lagon_runtime::isolate::{Isolate, IsolateOptions};
 use lagon_runtime::runtime::{Runtime, RuntimeOptions};
 use lazy_static::lazy_static;
-use log::error;
+use log::{error, info};
+use lru_time_cache::LruCache;
 use metrics::increment_counter;
 use metrics_exporter_prometheus::PrometheusBuilder;
 use mysql::{Opts, Pool};
@@ -22,6 +23,7 @@ use std::collections::HashMap;
 use std::convert::Infallible;
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::RwLock;
 use tokio_util::task::LocalPoolHandle;
 
@@ -35,9 +37,15 @@ mod deployments;
 mod logger;
 
 lazy_static! {
-    static ref ISOLATES: RwLock<HashMap<usize, HashMap<String, Isolate>>> =
+    static ref ISOLATES: RwLock<HashMap<usize, LruCache<String, Isolate>>> =
         RwLock::new(HashMap::new());
     static ref X_FORWARDED_FOR: String = String::from("X-Forwarded-For");
+    static ref ISOLATES_CACHE_SECONDS: Duration = Duration::from_secs(
+        dotenv::var("LAGON_ISOLATES_CACHE_SECONDS")
+            .expect("LAGON_ISOLATES_CACHE_SECONDS must be set")
+            .parse()
+            .expect("Failed to parse LAGON_ISOLATES_CACHE_SECONDS")
+    );
 }
 
 const POOL_SIZE: usize = 8;
@@ -116,18 +124,24 @@ async fn handle_request(
                             // deployment and that the isolate should be called.
                             // TODO: read() then write() if not present
                             let mut isolates = ISOLATES.write().await;
-                            let thread_isolates =
-                                isolates.entry(thread_id).or_insert_with(HashMap::new);
+                            let thread_isolates = isolates.entry(thread_id).or_insert_with(|| {
+                                LruCache::with_expiry_duration(*ISOLATES_CACHE_SECONDS)
+                            });
 
                             let isolate = thread_isolates.entry(hostname).or_insert_with(|| {
+                                info!("Creating new isolate: {} ", deployment.id);
                                 // TODO: handle read error
                                 let code = get_deployment_code(deployment).unwrap();
                                 let options = IsolateOptions::new(code)
                                     .with_environment_variables(
                                         deployment.environment_variables.clone(),
                                     )
                                     .with_memory(deployment.memory)
-                                    .with_timeout(deployment.timeout);
+                                    .with_timeout(deployment.timeout)
+                                    .with_id(deployment.id.clone())
+                                    .with_on_drop_callback(Box::new(|id| {
+                                        info!("Dropping isolate: {}", id.unwrap());
+                                    }));
 
                                 Isolate::new(options)
                             });