feat(hermes): Add thread pool for wasm modules parallel execution (#488)

Author: aido-mth

.config/dictionaries/project.dic

@@ -1,6 +1,9 @@
 aarch
 aido
 asat
+asyncio
+auditability
+autocheckpoint
 backpressure
 bkioshn
 blockfetch
@@ -74,6 +77,11 @@ mkdelay
 mkdirat
 nanos
 nextest
+nolfs
+nomutex
+nostack
+notadb
+nsec
 oneshot
 openat
 outlen
@@ -122,6 +130,7 @@ testdocs
 testunit
 thiserror
 thollander
+threadsafe
 timelike
 tinygo
 toolsets

hermes/Cargo.lock

@@ -2,7 +2,8 @@
 resolver = "2"
 members = [
     "bin",
-    "bin/tests/integration/components/http_request_rte_01"
+    "bin/tests/integration/components/http_request_rte_01",
+    "bin/tests/integration/components/sleep_component",
 ]
 default-members = [
     "bin",

hermes/Cargo.toml

@@ -66,6 +66,7 @@ chrono = { version = "0.4.41", features = ["now"] }
 chrono-tz = "0.10.4"
 saffron = "0.1.0"
 tokio = { version = "1.47.1", features = ["macros", "sync", "rt-multi-thread", "rt", "net"] }
+rayon = "1.11.0"
 libsqlite3-sys = {version="0.35.0", features = ["bundled"] }
 stringzilla = "3.12.6"
 temp-dir = "0.1.16"

hermes/bin/Cargo.toml

@@ -4,6 +4,7 @@ use std::{collections::HashMap, sync::Arc};
 
 use crate::{
     event::HermesEventPayload,
+    pool,
     runtime_context::HermesRuntimeContext,
     runtime_extensions::{
         init::trait_event::{RteEvent, RteInitEvent},
@@ -41,7 +42,7 @@ pub(crate) struct Application {
     name: ApplicationName,
 
     /// WASM modules
-    indexed_modules: HashMap<ModuleId, Module>,
+    indexed_modules: HashMap<ModuleId, Arc<Module>>,
 
     /// Application's `Vfs` instance
     vfs: Arc<Vfs>,
@@ -57,7 +58,7 @@ impl Application {
     ) -> Self {
         let indexed_modules = modules
             .into_iter()
-            .map(|module| (module.id().clone(), module))
+            .map(|module| (module.id().clone(), Arc::new(module)))
             .collect();
         Self {
             name: app_name,
@@ -79,66 +80,78 @@ impl Application {
     /// Dispatch event for all available modules.
     pub(crate) fn dispatch_event(
         &self,
-        event: &dyn HermesEventPayload,
-    ) -> anyhow::Result<()> {
+        event: &Arc<dyn HermesEventPayload>,
+    ) {
         for module in self.indexed_modules.values() {
             module_dispatch_event(
-                module,
+                module.clone(),
                 self.name.clone(),
                 module.id().clone(),
                 self.vfs.clone(),
-                event,
-            )?;
+                event.clone(),
+            );
         }
-        Ok(())
     }
 
     /// Dispatch event for the target module by the `module_id`.
     pub(crate) fn dispatch_event_for_target_module(
         &self,
         module_id: ModuleId,
-        event: &dyn HermesEventPayload,
+        event: Arc<dyn HermesEventPayload>,
     ) -> anyhow::Result<()> {
         let module = self
             .indexed_modules
             .get(&module_id)
             .ok_or(anyhow::anyhow!("Module {module_id} not found"))?;
         module_dispatch_event(
-            module,
+            module.clone(),
             self.name.clone(),
             module_id,
             self.vfs.clone(),
             event,
-        )
+        );
+        Ok(())
     }
 }
 
 /// Dispatch event
 pub(crate) fn module_dispatch_event(
-    module: &Module,
+    module: Arc<Module>,
     app_name: ApplicationName,
     module_id: ModuleId,
     vfs: Arc<Vfs>,
-    event: &dyn HermesEventPayload,
-) -> anyhow::Result<()> {
-    let runtime_ctx = HermesRuntimeContext::new(
-        app_name,
-        module_id,
-        event.event_name().to_string(),
-        module.exec_counter(),
-        vfs,
-    );
-
-    // Advise Runtime Extensions of a new context
-    // TODO: Better handle errors.
-    RteEvent::new().init(&runtime_ctx)?;
-    // TODO: (SJ) Remove when all RTE's are migrated.
-    new_context(&runtime_ctx);
-
-    module.execute_event(event, runtime_ctx.clone())?;
-
-    // Advise Runtime Extensions that context can be cleaned up.
-    RteEvent::new().fini(&runtime_ctx)?;
-
-    Ok(())
+    event: Arc<dyn HermesEventPayload>,
+) {
+    // TODO(@aido-mth): fix how init is processed. https://github.com/input-output-hk/hermes/issues/490
+    pool::execute(move || {
+        let runtime_ctx = HermesRuntimeContext::new(
+            app_name,
+            module_id,
+            event.event_name().to_string(),
+            module.exec_counter(),
+            vfs,
+        );
+
+        // Advise Runtime Extensions of a new context
+        // TODO: Better handle errors.
+        if let Err(err) = RteEvent::new().init(&runtime_ctx) {
+            tracing::error!("module event initialization failed: {err}");
+            return;
+        }
+
+        // TODO: (SJ) Remove when all RTE's are migrated.
+        new_context(&runtime_ctx);
+
+        if let Err(err) = module.execute_event(event.as_ref(), runtime_ctx.clone()) {
+            tracing::error!("module event execution failed: {err}");
+            return;
+        }
+
+        // Advise Runtime Extensions that context can be cleaned up.
+        drop(
+            RteEvent::new()
+                .fini(&runtime_ctx)
+                .inspect_err(|err| tracing::error!("module event finalization failed: {err}")),
+        );
+    });
 }

hermes/bin/src/app.rs

@@ -13,7 +13,7 @@ use crate::{
         app::{build_app, ApplicationPackage},
         sign::certificate::{self, Certificate},
     },
-    reactor,
+    pool, reactor,
 };
 
 /// Run cli command
@@ -56,6 +56,7 @@ impl Run {
         let app = build_app(&package, hermes_home_dir)?;
 
         let exit_lock = reactor::init()?;
+        pool::init()?;
         println!(
             "{} Loading application {}...",
             Emoji::new("🛠️", ""),
@@ -69,6 +70,8 @@ impl Run {
             exit_lock.wait()
         };
 
+        // Wait for scheduled tasks to be finished.
+        pool::terminate();
         Ok(exit)
     }
 }

hermes/bin/src/cli/run.rs

@@ -2,6 +2,8 @@
 
 pub mod queue;
 
+use std::sync::Arc;
+
 use crate::{
     app::ApplicationName,
     wasm::module::{ModuleId, ModuleInstance},
@@ -47,7 +49,7 @@ pub(crate) enum TargetModule {
 /// Hermes event
 pub(crate) struct HermesEvent {
     /// The payload carried by the `HermesEvent`.
-    payload: Box<dyn HermesEventPayload>,
+    payload: Arc<dyn HermesEventPayload>,
 
     /// Target app
     target_app: TargetApp,
@@ -64,15 +66,15 @@ impl HermesEvent {
         target_module: TargetModule,
     ) -> Self {
         Self {
-            payload: Box::new(payload),
+            payload: Arc::new(payload),
             target_app,
             target_module,
         }
     }
 
     /// Get event's payload
-    pub(crate) fn payload(&self) -> &dyn HermesEventPayload {
-        self.payload.as_ref()
+    pub(crate) fn payload(&self) -> &Arc<dyn HermesEventPayload> {
+        &self.payload
     }
 
     /// Get event's target app

hermes/bin/src/event/mod.rs

@@ -96,15 +96,14 @@ fn targeted_module_event_execution(
 
     match event.target_module() {
         TargetModule::All => {
-            if let Err(err) = app.dispatch_event(event.payload()) {
-                tracing::error!("{err}");
-            }
+            app.dispatch_event(event.payload());
         },
         TargetModule::List(target_modules) => {
             for target_module_id in target_modules {
-                if let Err(err) =
-                    app.dispatch_event_for_target_module(target_module_id.clone(), event.payload())
-                {
+                if let Err(err) = app.dispatch_event_for_target_module(
+                    target_module_id.clone(),
+                    event.payload().clone(),
+                ) {
                     tracing::error!("{err}");
                 }
             }

hermes/bin/src/event/queue.rs

@@ -10,6 +10,7 @@ pub mod hdf5;
 pub mod ipfs;
 pub mod logger;
 pub mod packaging;
+pub mod pool;
 pub mod reactor;
 pub mod runtime_context;
 pub mod runtime_extensions;

hermes/bin/src/lib.rs

@@ -8,6 +8,7 @@ mod hdf5;
 mod ipfs;
 mod logger;
 mod packaging;
+mod pool;
 mod reactor;
 mod runtime_context;
 mod runtime_extensions;