Added #actor macro to register the actor (#50)

* updated generated api clients

* clippy

* updated api to get list of registerd apis

* wip ui

* added how_to.md

* added register actor macro

Author: satyamjay-iitd

Cargo.lock

@@ -20,6 +20,9 @@ tracing = {version = "0.1.41", features = ["std", "attributes"] }
 tracing-subscriber = "0.3.19"
 tracing-shared = "0.1.5"
 criterion = "0.6.0"
+serde_json = "1.0.143"
+inventory = "0.3.21"
+reactor-macros = { path = "../macros" }
 
 [dev-dependencies]
 lazy_static = "1.5.0"

actor/Cargo.toml

@@ -1,10 +1,11 @@
 // #![feature(log_syntax)]
 
-use std::{borrow::Cow, marker::PhantomData};
+use std::{borrow::Cow, collections::HashMap, marker::PhantomData};
 
 use bincode::{Decode, Encode};
 use err::ActorError;
 use futures::future::join_all;
+pub use inventory as __inventory;
 use reactor_channel::{ReactorChannelTx, reactor_channel};
 use recv::rx;
 use send::tx;
@@ -24,6 +25,35 @@ mod send;
 
 pub use node_comm::{Connection, ControlInst, ControlReq, NodeComm};
 pub use reactor_channel::{HasPriority, MAX_PRIO};
+pub use reactor_macros::actor;
+
+pub type ActorSpawnCB = fn(RuntimeCtx, HashMap<String, serde_json::Value>);
+
+pub struct ExportedFn {
+    pub name: &'static str,
+    pub func: ActorSpawnCB,
+}
+
+__inventory::collect!(ExportedFn);
+
+#[macro_export]
+macro_rules! register_actor {
+    ($fn_name:path) => {
+        $crate::__inventory::submit! {
+            $crate::ExportedFn {
+                name: stringify!($fn_name),
+                func: $fn_name,
+            }
+        }
+    };
+}
+
+#[unsafe(no_mangle)]
+fn get_registered() -> Vec<String> {
+    inventory::iter::<ExportedFn>()
+        .map(|x| x.name.to_string())
+        .collect()
+}
 
 static CHANNEL_SIZE: usize = 1 << 20;
 /// Messages that can flow between the actors.

actor/src/lib.rs

@@ -0,0 +1,58 @@
+# How to use Reactor.
+
+
+## Initial Setup
+
+1. Start new Rust project:- `cargo init --lib my_proj`
+2. Modify `Cargo.toml` to contain
+```toml
+[lib]
+crate-type = ["cdylib", "rlib"] 
+```
+3. Add reactor as the dependency.
+`cargo add --git https://github.com/satyamjay-iitd/reactor/ reactor-actor`
+4. Define some actors.
+5. Compile the library:-
+```bash
+cargo build
+# You should now have a `target/debug/libmy_proj.so`, and ready to run a job.
+```
+
+
+## Running a job
+
+1. Install node controller.
+`cargo install --git https://github.com/satyamjay-iitd/reactor/ reactor_nctrl`
+2. Install job controller.
+`cargo install --git https://github.com/satyamjay-iitd/reactor/ reactor_jctrl`
+3. Start the node controller
+```bash
+# Note that the directory must contain the *.so file.
+reactor_nctrl --port 3000 ./target/debug`
+```
+4. Define job in a toml file.
+```toml
+[[ops]]
+name = "pingpong"
+lib_name = "ping_pong_actor"
+
+[[nodes]]
+name = "node1"
+hostname = "0.0.0.0"
+port = 3000
+
+[placement]
+  [[placement.pingpong]]
+  nodename = "node1"
+  actor_name = "pinger"
+  other = "ponger"
+
+  [[placement.pingpong]]
+  nodename = "node1"
+  actor_name = "ponger"
+  other = "pinger"
+```
+5. Run the job.
+```bash
+reactor_jctrl ./job.toml
+```

examples/how_to.md

@@ -3,7 +3,7 @@ pub use reactor_actor::setup_shared_logger_ref;
 use bincode::{Decode, Encode};
 
 use reactor_actor::codec::BincodeCodec;
-use reactor_actor::{BehaviourBuilder, RouteTo, RuntimeCtx};
+use reactor_actor::{BehaviourBuilder, RouteTo, RuntimeCtx, actor};
 use reactor_macros::{DefaultPrio, Msg as DeriveMsg};
 use std::collections::HashMap;
 use std::time::Duration;
@@ -77,27 +77,25 @@ impl Sender {
 //                                ACTORS
 // //////////////////////////////////////////////////////////////////////////////
 
-pub async fn actor(ctx: RuntimeCtx, other_addr: String) {
-    BehaviourBuilder::new(Processor {}, BincodeCodec::default())
-        .send(Sender::new(other_addr))
-        .generator_if(ctx.addr == "pinger", || vec![PingPongMsg::Ping].into_iter())
-        .build()
-        .run(ctx)
-        .await
-        .unwrap();
-}
-
 lazy_static::lazy_static! {
     static ref RUNTIME: tokio::runtime::Runtime = tokio::runtime::Runtime::new().unwrap();
 }
 
-#[unsafe(no_mangle)]
+#[actor]
 pub fn pingpong(ctx: RuntimeCtx, mut payload: HashMap<String, serde_json::Value>) {
-    let other: String = payload
+    let other_addr: String = payload
         .remove("other")
         .unwrap()
         .as_str()
         .unwrap()
         .to_string();
-    RUNTIME.spawn(actor(ctx, other));
+    RUNTIME.spawn(async move {
+        BehaviourBuilder::new(Processor {}, BincodeCodec::default())
+            .send(Sender::new(other_addr))
+            .generator_if(ctx.addr == "pinger", || vec![PingPongMsg::Ping].into_iter())
+            .build()
+            .run(ctx)
+            .await
+            .unwrap();
+    });
 }

examples/ping_pong/src/lib.rs

@@ -11,7 +11,7 @@ use writer::WriterIn;
 use crate::reader::{ReadAck, ReadOut, reader as reader_behaviour};
 use crate::server::server as server_behaviour;
 use crate::writer::{WriteAck, WriteOut, writer as writer_behaviour};
-use reactor_actor::RuntimeCtx;
+use reactor_actor::{RuntimeCtx, actor};
 use reactor_macros::msg_converter;
 use std::collections::HashMap;
 // //////////////////////////////////////////////////////////////////////////////
@@ -34,12 +34,12 @@ lazy_static::lazy_static! {
     static ref RUNTIME: tokio::runtime::Runtime = tokio::runtime::Runtime::new().unwrap();
 }
 
-#[unsafe(no_mangle)]
+#[actor]
 fn server(ctx: RuntimeCtx, _payload: HashMap<String, serde_json::Value>) {
     RUNTIME.spawn(server_behaviour(ctx, server_decoder));
 }
 
-#[unsafe(no_mangle)]
+#[actor]
 fn writer(ctx: RuntimeCtx, mut payload: HashMap<String, serde_json::Value>) {
     let server_addr = payload
         .remove("server_addr")
@@ -50,7 +50,7 @@ fn writer(ctx: RuntimeCtx, mut payload: HashMap<String, serde_json::Value>) {
     RUNTIME.spawn(writer_behaviour(ctx, server_addr, writer_decoder));
 }
 
-#[unsafe(no_mangle)]
+#[actor]
 fn reader(ctx: RuntimeCtx, mut payload: HashMap<String, serde_json::Value>) {
     let server_addr = payload
         .remove("server_addr")

examples/read_write_server/src/lib.rs

@@ -1,13 +1,27 @@
 use proc_macro::{self, TokenStream};
 use proc_macro2::TokenStream as TokenStream2;
 use quote::quote;
+use syn::ItemFn;
 use syn::{
     DeriveInput, Ident, Result, Token, Type, bracketed,
     parse::{Parse, ParseStream},
     parse_macro_input,
     punctuated::Punctuated,
 };
 
+#[proc_macro_attribute]
+pub fn actor(_attr: TokenStream, item: TokenStream) -> TokenStream {
+    let input = parse_macro_input!(item as ItemFn);
+    let name = &input.sig.ident;
+    let fn_tokens = quote! {
+        #[unsafe(no_mangle)]
+        #input
+
+        reactor_actor::register_actor!(#name);
+    };
+    fn_tokens.into()
+}
+
 #[proc_macro_derive(DefaultPrio)]
 pub fn auto_default_priority(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);

macros/src/lib.rs

@@ -29,7 +29,7 @@ use rpc::webserver;
 mod op_lib_manager;
 
 pub type NodeAddr = &'static str;
-pub type ActorSpawnCB = fn(RuntimeCtx, HashMap<String, serde_json::Value>);
+// pub type ActorSpawnCB = fn(RuntimeCtx, HashMap<String, serde_json::Value>);
 
 pub type SetupSharedLogger = fn(SharedLogger);
 
@@ -48,7 +48,7 @@ pub(crate) struct RegisterResult {}
 #[derive(Debug)]
 pub(crate) struct NodeStatus {
     actors: Vec<String>,
-    loaded_libs: Vec<String>,
+    loaded_libs: HashMap<String, Vec<String>>,
 }
 
 /// Global Controller
@@ -201,6 +201,8 @@ async fn handle_job_req(
 
             let lib = op_lib.get_lib(&lib_name);
             unsafe {
+                use reactor_actor::ActorSpawnCB;
+
                 let shared_logger: libloading::Symbol<SetupSharedLogger> =
                     lib.get(b"setup_shared_logger_ref").unwrap();
                 let logger = SharedLogger::new();

node/src/lib.rs

@@ -6,23 +6,35 @@ use crate::LibName;
 
 #[derive(Default, Debug)]
 pub(crate) struct OpLibrary {
-    container: HashMap<LibName, Library>,
+    container: HashMap<LibName, (Library, Vec<String>)>,
 }
 
 impl OpLibrary {
     pub(crate) fn add_lib(&mut self, name: LibName, library: Library) {
-        self.container.insert(name, library);
+        let registered = unsafe {
+            if let Ok(get_registered) =
+                library.get::<libloading::Symbol<fn() -> Vec<String>>>(b"get_registered")
+            {
+                get_registered()
+            } else {
+                return;
+            }
+        };
+        self.container.insert(name, (library, registered));
     }
 
     pub(crate) fn get_lib(&self, lib_name: &str) -> &Library {
-        self.container.get(lib_name).unwrap()
+        &self.container.get(lib_name).unwrap().0
     }
 
     pub(crate) fn num_libs(&self) -> usize {
         self.container.len()
     }
 
-    pub(crate) fn lib_names(&self) -> Vec<LibName> {
-        self.container.keys().cloned().collect()
+    pub(crate) fn lib_names(&self) -> HashMap<LibName, Vec<String>> {
+        self.container
+            .iter()
+            .map(|(name, (_, ops))| (name.clone(), ops.clone()))
+            .collect()
     }
 }

node/src/op_lib_manager.rs

@@ -215,7 +215,7 @@ async fn stop_all_actors(State(state): State<Arc<AppState>>) -> impl IntoRespons
 #[derive(Serialize, ToSchema)]
 struct StatusResponse {
     actors: Vec<String>,
-    loaded_libs: Vec<String>,
+    loaded_libs: HashMap<String, Vec<String>>,
 }
 #[cfg_attr(feature="swagger", utoipa::path(
     get,