feat: Hermes event queue and Hermes reactor

hermes/Cargo.toml

@@ -59,5 +59,10 @@ anyhow = "1.0.71"
 blake2b_simd = "1.0.2"
 hex-literal = "0.4.1"
 thiserror = "1.0.56"
+tokio = "1.34.0"
+hex = "0.4.3"
+tracing = "0.1.40"
+tracing-subscriber = "0.3.18"
 criterion = "0.5.1"
 time = "0.3.34"
+once_cell = "1.19.0"

hermes/bin/Cargo.toml

@@ -29,3 +29,4 @@ hex-literal = { workspace = true }
 thiserror = { workspace = true }
 criterion = { workspace = true, optional = true }
 time = { workspace = true }
+once_cell = { workspace = true }

hermes/bin/src/app.rs

@@ -0,0 +1,47 @@
+//! Hermes app implementation.
+
+use std::collections::HashMap;
+
+use crate::wasm::module::{Module, ModuleId};
+
+/// Hermes App Name type
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) struct HermesAppName(pub(crate) String);
+
+/// Convenient type alias for indexed apps map (`HermesAppName` -> `HermesApp`)
+pub(crate) type IndexedApps = HashMap<HermesAppName, HermesApp>;
+
+/// Hermes app
+pub(crate) struct HermesApp {
+    /// App name
+    app_name: HermesAppName,
+
+    /// WASM modules
+    indexed_modules: HashMap<ModuleId, Module>,
+}
+
+impl HermesApp {
+    /// Create a new Hermes app
+    #[allow(dead_code)]
+    pub(crate) fn new(app_name: HermesAppName, module_bytes: Vec<Vec<u8>>) -> anyhow::Result<Self> {
+        let mut modules = HashMap::with_capacity(module_bytes.len());
+        for module_bytes in module_bytes {
+            let module = Module::new(&module_bytes)?;
+            modules.insert(module.id().clone(), module);
+        }
+        Ok(Self {
+            app_name,
+            indexed_modules: modules,
+        })
+    }
+
+    /// Get app name
+    pub(crate) fn app_name(&self) -> &HermesAppName {
+        &self.app_name
+    }
+
+    /// Get indexed modules
+    pub(crate) fn indexed_modules(&self) -> &HashMap<ModuleId, Module> {
+        &self.indexed_modules
+    }
+}

hermes/bin/src/event/mod.rs

@@ -0,0 +1,81 @@
+//! Hermes event's primitives.
+
+pub(crate) mod queue;
+
+use crate::{
+    app::HermesAppName,
+    wasm::module::{ModuleId, ModuleInstance},
+};
+
+/// A trait for defining the behavior of a Hermes event.
+pub(crate) trait HermesEventPayload: Send + Sync + 'static {
+    /// Returns the name of the event associated with the payload.
+    fn event_name(&self) -> &str;
+
+    /// Executes the behavior associated with the payload, using the provided executor.
+    ///
+    /// # Arguments
+    ///
+    /// * `executor` - The executor to use for executing the payload's behavior.
+    ///
+    /// # Returns
+    ///
+    /// An `anyhow::Result` indicating the success or failure of the payload execution.
+    fn execute(&self, module: &mut ModuleInstance) -> anyhow::Result<()>;
+}
+
+/// Target Hermes app to execute the event
+pub(crate) enum TargetApp {
+    /// Execute for all available apps
+    _All,
+    /// Execute for a specific list of apps
+    List(Vec<HermesAppName>),
+}
+
+/// Target WASM module to execute the event
+pub(crate) enum TargetModule {
+    /// Execute for all available modules
+    All,
+    /// Execute for a specific list of modules
+    _List(Vec<ModuleId>),
+}
+
+/// Hermes event
+pub(crate) struct HermesEvent {
+    /// The payload carried by the HermesEvent.
+    payload: Box<dyn HermesEventPayload>,
+
+    /// Target app
+    target_app: TargetApp,
+
+    /// Target module
+    target_module: TargetModule,
+}
+
+impl HermesEvent {
+    /// Create a new Hermes event
+    pub(crate) fn new(
+        payload: impl HermesEventPayload, target_app: TargetApp, target_module: TargetModule,
+    ) -> Self {
+        Self {
+            payload: Box::new(payload),
+            target_app,
+            target_module,
+        }
+    }
+
+    /// Get event's payload
+    pub(crate) fn payload(&self) -> &dyn HermesEventPayload {
+        self.payload.as_ref()
+    }
+
+    /// Get event's target app
+    pub(crate) fn target_app(&self) -> &TargetApp {
+        &self.target_app
+    }
+
+    /// Get event's target module
+    pub(crate) fn target_module(&self) -> &TargetModule {
+        &self.target_module
+    }
+}

hermes/bin/src/event/queue.rs

@@ -0,0 +1,197 @@
+//! Hermes event queue implementation.
+
+use std::{
+    sync::{
+        mpsc::{Receiver, Sender},
+        Arc,
+    },
+    thread::{self, JoinHandle},
+};
+
+use once_cell::sync::OnceCell;
+
+use super::{HermesEvent, HermesEventPayload, TargetApp, TargetModule};
+use crate::{
+    app::{HermesAppName, IndexedApps},
+    runtime_context::HermesRuntimeContext,
+    runtime_extensions::new_context,
+    wasm::module::{Module, ModuleId},
+};
+
+/// Singleton instance of the Hermes event queue.
+static EVENT_QUEUE_INSTANCE: OnceCell<HermesEventQueue> = OnceCell::new();
+
+/// Hermes event queue error
+#[derive(thiserror::Error, Debug, Clone)]
+pub(crate) enum Error {
+    /// Target app not found.
+    #[error("Target app not found, app name: {0:?}.")]
+    AppNotFound(HermesAppName),
+
+    /// Target module not found.
+    #[error("Target module not found, module id: {0:?}.")]
+    ModuleNotFound(ModuleId),
+
+    /// Failed to add event into the event queue. Event queue is closed.
+    #[error("Failed to add event into the event queue. Event queue is closed.")]
+    CannotAddEvent,
+
+    /// Failed when event queue already been initialized.
+    #[error("Event queue already been initialized.")]
+    AlreadyInitialized,
+
+    /// Failed when event queue not been initialized.
+    #[error("Event queue not been initialized. Call `HermesEventQueue::init` first.")]
+    NotInitialized,
+
+    /// Event loop has crashed unexpectedly.
+    #[error("Event loop has crashed unexpectedly.")]
+    EventLoopPanics,
+}
+
+/// Hermes event queue.
+/// It is a singleton struct.
+struct HermesEventQueue {
+    /// Hermes event queue sender
+    sender: Sender<HermesEvent>,
+}
+
+/// Hermes event queue execution loop thread handler
+pub(crate) struct HermesEventLoopHandler {
+    /// Hermes event queue execution loop thread handler
+    handle: Option<JoinHandle<anyhow::Result<()>>>,
+}
+
+impl HermesEventLoopHandler {
+    /// Join the event loop thread
+    pub(crate) fn join(&mut self) -> anyhow::Result<()> {
+        match self.handle.take() {
+            Some(handle) => handle.join().map_err(|_| Error::EventLoopPanics)?,
+            None => Ok(()),
+        }
+    }
+}
+
+/// Creates a new instance of the `HermesEventQueue`.
+/// Runs an event loop thread.
+///
+/// # Errors:
+/// - `Error::AlreadyInitialized`
+pub(crate) fn init(indexed_apps: Arc<IndexedApps>) -> anyhow::Result<HermesEventLoopHandler> {
+    let (sender, receiver) = std::sync::mpsc::channel();
+
+    EVENT_QUEUE_INSTANCE
+        .set(HermesEventQueue { sender })
+        .map_err(|_| Error::AlreadyInitialized)?;
+
+    Ok(HermesEventLoopHandler {
+        handle: Some(thread::spawn(move || {
+            event_execution_loop(&indexed_apps, receiver)
+        })),
+    })
+}
+
+/// Add event into the event queue
+///
+/// # Errors:
+/// - `Error::CannotAddEvent`
+/// - `Error::NotInitialized`
+pub(crate) fn send(event: HermesEvent) -> anyhow::Result<()> {
+    let queue = EVENT_QUEUE_INSTANCE.get().ok_or(Error::NotInitialized)?;
+
+    queue
+        .sender
+        .send(event)
+        .map_err(|_| Error::CannotAddEvent)?;
+    Ok(())
+}
+
+/// Execute a hermes event on the provided module and all necessary info.
+///
+/// # Errors:
+/// - `wasm::module::BadWASMModuleError`
+fn event_dispatch(
+    app_name: HermesAppName, module_id: ModuleId, event: &dyn HermesEventPayload, module: &Module,
+) -> anyhow::Result<()> {
+    let runtime_context = HermesRuntimeContext::new(
+        app_name,
+        module_id,
+        event.event_name().to_string(),
+        module.exec_counter(),
+    );
+
+    // Advise Runtime Extensions of a new context
+    new_context(&runtime_context);
+
+    module.execute_event(event, runtime_context)?;
+    Ok(())
+}
+
+/// Executes provided Hermes event filtering by target app and target module.
+///
+/// # Errors:
+/// - `Error::ModuleNotFound`
+/// - `Error::AppNotFound`
+/// - `wasm::module::BadWASMModuleError`
+fn targeted_event_execution(indexed_apps: &IndexedApps, event: &HermesEvent) -> anyhow::Result<()> {
+    // Gather target apps
+    let target_apps = match event.target_app() {
+        TargetApp::_All => indexed_apps.iter().collect(),
+        TargetApp::List(target_apps) => {
+            let mut res = Vec::new();
+            for app_name in target_apps {
+                let app = indexed_apps
+                    .get(app_name)
+                    .ok_or(Error::AppNotFound(app_name.to_owned()))?;
+                res.push((app_name, app));
+            }
+            res
+        },
+    };
+    // Gather target modules
+    let target_modules = match event.target_module() {
+        TargetModule::All => {
+            let mut res = Vec::new();
+            for (app_name, app) in target_apps {
+                for (module_id, module) in app.indexed_modules() {
+                    res.push((app_name, module_id, module));
+                }
+            }
+            res
+        },
+        TargetModule::_List(target_modules) => {
+            let mut res = Vec::new();
+            for (app_name, app) in target_apps {
+                for module_id in target_modules {
+                    let module = app
+                        .indexed_modules()
+                        .get(module_id)
+                        .ok_or(Error::ModuleNotFound(module_id.to_owned()))?;
+                    res.push((app_name, module_id, module));
+                }
+            }
+            res
+        },
+    };
+
+    // Event dispatch
+    for (app_name, module_id, module) in target_modules {
+        event_dispatch(app_name.clone(), module_id.clone(), event.payload(), module)?;
+    }
+    Ok(())
+}
+
+/// Executes Hermes events from the provided receiver .
+///
+/// # Errors:
+/// - `Error::ModuleNotFound`
+/// - `Error::AppNotFound`
+/// - `wasm::module::BadWASMModuleError`
+fn event_execution_loop(
+    indexed_apps: &IndexedApps, receiver: Receiver<HermesEvent>,
+) -> anyhow::Result<()> {
+    for event in receiver {
+        targeted_event_execution(indexed_apps, &event)?;
+    }
+    Ok(())
+}

hermes/bin/src/lib.rs

@@ -1,9 +1,11 @@
 //! Intentionally empty
 //! This file exists, so that doc tests can be used inside binary crates.
 
-mod event_queue;
+mod app;
+mod event;
+mod reactor;
+mod runtime_context;
 mod runtime_extensions;
-mod state;
 mod wasm;
 
 #[cfg(feature = "bench")]

hermes/bin/src/main.rs

@@ -1,8 +1,10 @@
 //! The Hermes Node.
 
-mod event_queue;
+mod app;
+mod event;
+mod reactor;
+mod runtime_context;
 mod runtime_extensions;
-mod state;
 mod wasm;
 
 #[cfg(feature = "bench")]