Backporting Memory Optimizations to 1.21.rc (#883)

* fix: Moving Arc<openrpc_state> Arc<exclusory> , Arc<ProviderRegistrations>.

* fix: Moving Arc<Manifests> Arc<rule_engine>.

* fix: Improving rule parse and error.

* fix: Avoid cloning Methods in rpc_router.

* feat: Adding memory logging feature.

Author: kvfasil

core/main/src/bootstrap/boot.rs

@@ -21,6 +21,7 @@ use ripple_sdk::{
         RippleResponse,
     },
     log::error,
+    utils::test_utils::log_memory_usage,
 };
 
 use crate::state::bootstrap_state::BootstrapState;
@@ -61,18 +62,31 @@ use super::{
 
 ///
 pub async fn boot(state: BootstrapState) -> RippleResponse {
+    log_memory_usage("boot-Begining");
     let bootstrap = Bootstrap::new(state);
+    log_memory_usage("After-LoggingBootstrapStep");
     execute_step(StartCommunicationBroker, &bootstrap).await?;
+    log_memory_usage("After-StartCommunicationBroker");
     execute_step(SetupExtnClientStep, &bootstrap).await?;
+    log_memory_usage("After-SetupExtnClientStep");
     execute_step(LoadExtensionMetadataStep, &bootstrap).await?;
+    log_memory_usage("After-LoadExtensionMetadataStep");
     execute_step(LoadExtensionsStep, &bootstrap).await?;
+    log_memory_usage("After-LoadExtensionsStep");
     execute_step(StartExtnChannelsStep, &bootstrap).await?;
+    log_memory_usage("After-StartExtnChannelsStep");
     execute_step(StartAppManagerStep, &bootstrap).await?;
+    log_memory_usage("After-StartAppManagerStep");
     execute_step(StartOtherBrokers, &bootstrap).await?;
+    log_memory_usage("After-StartOtherBrokers");
     execute_step(LoadDistributorValuesStep, &bootstrap).await?;
+    log_memory_usage("After-LoadDistributorValuesStep");
     execute_step(CheckLauncherStep, &bootstrap).await?;
+    log_memory_usage("After-CheckLauncherStep");
     execute_step(StartWsStep, &bootstrap).await?;
+    log_memory_usage("After-StartWsStep");
     execute_step(FireboltGatewayStep, &bootstrap).await?;
+    log_memory_usage("After-FireboltGatewayStep");
     Ok(())
 }
 

core/main/src/bootstrap/manifest/apps.rs

@@ -40,7 +40,7 @@ type AppLibraryLoader = Vec<fn() -> Result<(String, Vec<AppLibraryEntry>), Rippl
 
 fn try_app_library_files() -> Result<Vec<AppLibraryEntry>, RippleError> {
     let al_arr: AppLibraryLoader = if cfg!(feature = "local_dev") {
-        vec![load_from_env, load_from_home]
+        vec![load_from_env, load_from_home, load_from_etc]
     } else if cfg!(test) {
         vec![load_from_env]
     } else {

core/main/src/bootstrap/manifest/device.rs

@@ -36,7 +36,7 @@ type DeviceManifestLoader = Vec<fn() -> Result<(String, DeviceManifest), RippleE
 
 fn try_manifest_files() -> Result<DeviceManifest, RippleError> {
     let dm_arr: DeviceManifestLoader = if cfg!(feature = "local_dev") {
-        vec![load_from_env, load_from_home]
+        vec![load_from_env, load_from_home, load_from_etc]
     } else if cfg!(test) {
         vec![load_from_env]
     } else {

core/main/src/bootstrap/manifest/extn.rs

@@ -31,7 +31,7 @@ type ExtnManifestLoader = Vec<fn() -> Result<(String, ExtnManifest), RippleError
 
 fn try_manifest_files() -> Result<ExtnManifest, RippleError> {
     let dm_arr: ExtnManifestLoader = if cfg!(feature = "local_dev") {
-        vec![load_from_env, load_from_home]
+        vec![load_from_env, load_from_home, load_from_etc]
     } else if cfg!(test) {
         vec![load_from_env]
     } else {

core/main/src/broker/endpoint_broker.rs

@@ -371,7 +371,7 @@ pub struct EndpointBrokerState {
     callback: BrokerCallback,
     request_map: Arc<RwLock<HashMap<u64, BrokerRequest>>>,
     extension_request_map: Arc<RwLock<HashMap<u64, ExtnMessage>>>,
-    rule_engine: RuleEngine,
+    rule_engine: Arc<RwLock<RuleEngine>>,
     cleaner_list: Arc<RwLock<Vec<BrokerCleaner>>>,
     reconnect_tx: Sender<BrokerConnectRequest>,
     provider_broker_state: ProvideBrokerState,
@@ -384,7 +384,7 @@ impl Default for EndpointBrokerState {
             callback: BrokerCallback::default(),
             request_map: Arc::new(RwLock::new(HashMap::new())),
             extension_request_map: Arc::new(RwLock::new(HashMap::new())),
-            rule_engine: RuleEngine::default(),
+            rule_engine: Arc::new(RwLock::new(RuleEngine::default())),
             cleaner_list: Arc::new(RwLock::new(Vec::new())),
             reconnect_tx: mpsc::channel(2).0,
             provider_broker_state: ProvideBrokerState::default(),
@@ -406,7 +406,7 @@ impl EndpointBrokerState {
             callback: BrokerCallback { sender: tx },
             request_map: Arc::new(RwLock::new(HashMap::new())),
             extension_request_map: Arc::new(RwLock::new(HashMap::new())),
-            rule_engine,
+            rule_engine: Arc::new(RwLock::new(rule_engine)),
             cleaner_list: Arc::new(RwLock::new(Vec::new())),
             reconnect_tx,
             provider_broker_state: ProvideBrokerState::default(),
@@ -416,7 +416,7 @@ impl EndpointBrokerState {
         state
     }
     pub fn with_rules_engine(mut self, rule_engine: RuleEngine) -> Self {
-        self.rule_engine = rule_engine;
+        self.rule_engine = Arc::new(RwLock::new(rule_engine));
         self
     }
 
@@ -523,7 +523,16 @@ impl EndpointBrokerState {
         )
     }
     pub fn build_thunder_endpoint(&mut self) {
-        if let Some(endpoint) = self.rule_engine.rules.endpoints.get("thunder").cloned() {
+        let endpoint = {
+            self.rule_engine
+                .write()
+                .unwrap()
+                .rules
+                .endpoints
+                .get("thunder")
+                .cloned()
+        };
+        if let Some(endpoint) = endpoint {
             let request = BrokerConnectRequest::new(
                 "thunder".to_owned(),
                 endpoint.clone(),
@@ -534,7 +543,8 @@ impl EndpointBrokerState {
     }
 
     pub fn build_other_endpoints(&mut self, ps: PlatformState, session: Option<AccountSession>) {
-        for (key, endpoint) in self.rule_engine.rules.endpoints.clone() {
+        let endpoints = self.rule_engine.read().unwrap().rules.endpoints.clone();
+        for (key, endpoint) in endpoints {
             // skip thunder endpoint as it is already built using build_thunder_endpoint
             if let RuleEndpointProtocol::Thunder = endpoint.protocol {
                 continue;
@@ -719,7 +729,7 @@ impl EndpointBrokerState {
             rpc_request.ctx.clone(),
         )
         .emit_debug();
-        if let Some(rule) = self.rule_engine.get_rule(&rpc_request) {
+        if let Some(rule) = self.rule_engine.read().unwrap().get_rule(&rpc_request) {
             found_rule = Some(rule.clone());
 
             if let Some(endpoint) = rule.endpoint {
@@ -847,7 +857,7 @@ impl EndpointBrokerState {
     }
 
     pub fn get_rule(&self, rpc_request: &RpcRequest) -> Option<Rule> {
-        self.rule_engine.get_rule(rpc_request)
+        self.rule_engine.read().unwrap().get_rule(rpc_request)
     }
 
     // Method to cleanup all subscription on App termination

core/main/src/broker/rules_engine.rs

@@ -27,8 +27,12 @@ use ripple_sdk::{
 };
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::sync::{Mutex, MutexGuard, Once};
 use std::{fs, path::Path};
 
+static BASE_PARSE_CTX_INIT: Once = Once::new();
+static mut BASE_PARSE_CTX_PTR: Option<Mutex<ParseCtx>> = None;
+
 #[derive(Debug, Deserialize, Default, Clone)]
 pub struct RuleSet {
     pub endpoints: HashMap<String, RuleEndpoint>,
@@ -314,6 +318,24 @@ impl RuleEngine {
 /// let result = jq_compile(input, filter, String::new());
 /// assert_eq!(result.unwrap(), json!("SCXI11BEI_VBN_24Q2_sprint_20240620140024sdy_FG_GRT"));
 /// ```
+// Initializes the base ParseCtx with core and std filters, only once.
+fn get_parse_ctx() -> MutexGuard<'static, ParseCtx> {
+    BASE_PARSE_CTX_INIT.call_once(|| {
+        let mut ctx = ParseCtx::new(Vec::new());
+        ctx.insert_natives(jaq_core::core());
+        ctx.insert_defs(jaq_std::std());
+        unsafe {
+            BASE_PARSE_CTX_PTR = Some(Mutex::new(ctx));
+        }
+    });
+    unsafe {
+        BASE_PARSE_CTX_PTR
+            .as_ref()
+            .expect("BASE_PARSE_CTX_PTR not initialized")
+            .lock()
+            .expect("Failed to lock BASE_PARSE_CTX_PTR")
+    }
+}
 pub fn jq_compile(input: Value, filter: &str, reference: String) -> Result<Value, RippleError> {
     info!(
         "Jq rule {}  input {:?}, reference {}",
@@ -324,22 +346,22 @@ pub fn jq_compile(input: Value, filter: &str, reference: String) -> Result<Value
     // which do not include filters in the standard library
     // such as `map`, `select` etc.
 
-    let mut defs = ParseCtx::new(Vec::new());
-    defs.insert_natives(jaq_core::core());
-    defs.insert_defs(jaq_std::std());
     // parse the filter
     let (f, errs) = jaq_parse::parse(filter, jaq_parse::main());
     if !errs.is_empty() {
-        error!("Error in rule {:?}", errs);
+        error!("Error in rule {:?}: {:?}", reference, errs);
         return Err(RippleError::RuleError);
     }
+    // Lock and use the shared ParseCtx
+    let mut defs = get_parse_ctx();
     // compile the filter in the context of the given definitions
     let f = defs.compile(f.unwrap());
     if !defs.errs.is_empty() {
         error!("Error in rule {}", reference);
-        for (err, _) in defs.errs {
+        for (err, _) in &defs.errs {
             error!("reference={} {}", reference, err);
         }
+        defs.errs.clear(); // Clear errors before returning
         return Err(RippleError::RuleError);
     }
 

core/main/src/firebolt/handlers/device_rpc.rs

@@ -202,8 +202,9 @@ impl DeviceServer for DeviceImpl {
 
     async fn version(&self, ctx: CallContext) -> RpcResult<DeviceVersionResponse> {
         let firmware_info = self.firmware_info(ctx).await?;
-        let open_rpc_state = self.state.clone().open_rpc_state;
-        let api = open_rpc_state.get_open_rpc().info;
+        let open_rpc_state = &self.state.clone().open_rpc_state;
+        let api = &*open_rpc_state.get_open_rpc().clone();
+        let api = api.info.clone();
 
         // os is deprecated, for now senidng firmware ver in os as well
         let os_ver = firmware_info.clone().version;

core/main/src/firebolt/rpc_router.rs

@@ -17,13 +17,10 @@
 
 use futures::StreamExt;
 use jsonrpsee::{
-    core::{
-        server::{
-            helpers::MethodSink,
-            resource_limiting::Resources,
-            rpc_module::{MethodKind, Methods},
-        },
-        TEN_MB_SIZE_BYTES,
+    core::server::{
+        helpers::MethodSink,
+        resource_limiting::Resources,
+        rpc_module::{MethodCallback, MethodKind, Methods},
     },
     types::{error::ErrorCode, Id, Params},
 };
@@ -71,8 +68,12 @@ impl RouterState {
         let _ = methods_state.merge(methods.initialize_resources(&self.resources).unwrap());
     }
 
-    fn get_methods(&self) -> Methods {
-        self.methods.read().unwrap().clone()
+    pub fn get_method_entry(&self, method_name: &str) -> Option<(String, MethodCallback)> {
+        // Acquire a read lock without cloning the entire Methods registry
+        let methods_guard = self.methods.read().ok()?;
+        methods_guard
+            .method_with_name(method_name)
+            .map(|(name, method)| (name.to_owned(), method.clone()))
     }
 }
 
@@ -84,7 +85,7 @@ impl Default for RouterState {
 
 async fn resolve_route(
     platform_state: &mut PlatformState,
-    methods: Methods,
+    method_entry: Option<(String, MethodCallback)>,
     resources: Resources,
     req: RpcRequest,
 ) -> Result<ApiMessage, RippleError> {
@@ -93,26 +94,29 @@ async fn resolve_route(
     let request_c = req.clone();
     let sink_size = 1024 * 1024;
     let (sink_tx, mut sink_rx) = futures_channel::mpsc::unbounded::<String>();
-    let sink = MethodSink::new_with_limit(sink_tx, TEN_MB_SIZE_BYTES, 512 * 1024);
-    let method = request_c.method.clone();
+    let sink = MethodSink::new_with_limit(sink_tx, 1024 * 1024, 100 * 1024);
+    let method_name = request_c.method.clone();
 
     tokio::spawn(async move {
         let params_json = request_c.params_json.as_ref();
         let params = Params::new(Some(params_json));
 
-        match methods.method_with_name(&method) {
+        match method_entry {
             None => {
                 LogSignal::new(
                     "rpc_router".to_string(),
                     "resolve_route".into(),
                     request_c.clone(),
                 )
-                .with_diagnostic_context_item("error", &format!("Method not found: {}", method))
+                .with_diagnostic_context_item(
+                    "error",
+                    &format!("Method not found: {}", method_name),
+                )
                 .emit_error();
                 sink.send_error(id, ErrorCode::MethodNotFound.into());
             }
             Some((name, method)) => match &method.inner() {
-                MethodKind::Sync(callback) => match method.claim(name, &resources) {
+                MethodKind::Sync(callback) => match method.claim(&name, &resources) {
                     Ok(_guard) => {
                         if let Err(e) =
                             sink.send_raw((callback)(id.clone(), params, 512 * 1024).result)
@@ -134,7 +138,7 @@ async fn resolve_route(
                         sink.send_error(id, ErrorCode::MethodNotFound.into());
                     }
                 },
-                MethodKind::Async(callback) => match method.claim(name, &resources) {
+                MethodKind::Async(callback) => match method.claim(&name, &resources) {
                     Ok(guard) => {
                         let id = id.into_owned();
                         let params = params.into_owned();
@@ -209,7 +213,7 @@ impl RpcRouter {
         session: Session,
         timer: Option<Timer>,
     ) {
-        let methods = state.router_state.get_methods();
+        let method_entry = state.router_state.get_method_entry(&req.method);
         let resources = state.router_state.resources.clone();
 
         if let Some(overridden_method) = state.get_manifest().has_rpc_override_method(&req.method) {
@@ -218,7 +222,7 @@ impl RpcRouter {
         LogSignal::new("rpc_router".to_string(), "routing".into(), req.clone());
         tokio::spawn(async move {
             let start = Utc::now().timestamp_millis();
-            let resp = resolve_route(&mut state, methods, resources, req.clone()).await;
+            let resp = resolve_route(&mut state, method_entry, resources, req.clone()).await;
 
             let status = match resp.clone() {
                 Ok(msg) => {
@@ -247,7 +251,7 @@ impl RpcRouter {
         req: RpcRequest,
         extn_msg: ExtnMessage,
     ) {
-        let methods = state.router_state.get_methods();
+        let method_entry = state.router_state.get_method_entry(&req.method);
         let resources = state.router_state.resources.clone();
 
         let mut platform_state = state.clone();
@@ -258,7 +262,8 @@ impl RpcRouter {
         )
         .emit_debug();
         tokio::spawn(async move {
-            if let Ok(msg) = resolve_route(&mut platform_state, methods, resources, req).await {
+            if let Ok(msg) = resolve_route(&mut platform_state, method_entry, resources, req).await
+            {
                 return_extn_response(msg, extn_msg);
             }
         });

core/main/src/processor/config_processor.rs

@@ -135,7 +135,8 @@ impl ExtnRequestProcessor for ConfigRequestProcessor {
                     .unwrap_or_default(),
             ),
             Config::Firebolt => ExtnResponse::Value(
-                serde_json::to_value(state.open_rpc_state.get_open_rpc()).unwrap_or_default(),
+                serde_json::to_value(&*state.open_rpc_state.get_open_rpc().clone())
+                    .unwrap_or_default(),
             ),
             Config::RFC(flag) => {
                 let mut resp =