Add concurrent verb execution load tests

Spams a running server with brute force verb executions in an attempt to measure throughput.

bacon.toml

@@ -11,6 +11,7 @@ default_job = "daemon"
 command = [
     "cargo",
     "run",
+    "--release",
     "-p",
     "moor-daemon",
     "--",

crates/testing/load-tools/Cargo.toml

@@ -15,6 +15,10 @@ description = "Load testing and transaction model checking"
 name = "moor-model-checker"
 path = "src/tx-list-append.rs"
 
+[[bin]]
+name = "load-test"
+path = "src/verb-dispatch-load-test.rs"
+
 [dependencies]
 moor-values = { path = "../../common" }
 rpc-async-client = { path = "../../rpc/rpc-async-client" }

crates/testing/load-tools/src/setup.rs

@@ -12,24 +12,26 @@
 // this program. If not, see <https://www.gnu.org/licenses/>.
 //
 
-use eyre::bail;
+use eyre::{anyhow, bail};
 use moor_values::model::ObjectRef;
 use moor_values::tasks::VerbProgramError;
-use moor_values::{Obj, Symbol, SYSTEM_OBJECT};
-use rpc_async_client::pubsub_client::broadcast_recv;
+use moor_values::{Obj, Symbol, Var, SYSTEM_OBJECT};
+use rpc_async_client::pubsub_client::{broadcast_recv, events_recv};
 use rpc_async_client::rpc_client::RpcSendClient;
 use rpc_async_client::{ListenersClient, ListenersMessage};
 use rpc_common::HostClientToDaemonMessage::ConnectionEstablish;
 use rpc_common::{
-    AuthToken, ClientToken, ClientsBroadcastEvent, DaemonToClientReply, HostClientToDaemonMessage,
-    HostType, ReplyResult, VerbProgramResponse, CLIENT_BROADCAST_TOPIC,
+    AuthToken, ClientEvent, ClientToken, ClientsBroadcastEvent, DaemonToClientReply,
+    HostClientToDaemonMessage, HostType, ReplyResult, VerbProgramResponse, CLIENT_BROADCAST_TOPIC,
 };
+use std::collections::HashMap;
 use std::net::{Ipv4Addr, SocketAddr};
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
-use std::time::SystemTime;
+use std::time::{Instant, SystemTime};
 use tmq::subscribe::Subscribe;
 use tmq::{request, subscribe};
+use tokio::sync::Mutex;
 use tokio::task::JoinHandle;
 use tracing::{debug, error, info};
 use uuid::Uuid;
@@ -301,6 +303,7 @@ pub async fn initialization_session(
     info!("Initialization script executed successfully");
 
     for (verb_name, verb_code) in verbs {
+        info!("Compiling {} verb", verb_name);
         compile(
             &mut rpc_client,
             client_id,
@@ -321,6 +324,47 @@ pub async fn initialization_session(
 }
 
 pub struct ExecutionContext {
-    pub(crate) zmq_ctx: tmq::Context,
-    pub(crate) kill_switch: Arc<std::sync::atomic::AtomicBool>,
+    pub zmq_ctx: tmq::Context,
+    pub kill_switch: Arc<std::sync::atomic::AtomicBool>,
+}
+
+pub async fn listen_responses(
+    client_id: Uuid,
+    mut events_sub: Subscribe,
+    ks: Arc<AtomicBool>,
+    event_listen_task_results: Arc<Mutex<HashMap<usize, Result<Var, eyre::Error>>>>,
+) {
+    tokio::spawn(async move {
+        let start_time = Instant::now();
+        info!("Waiting for events...");
+        loop {
+            if ks.load(std::sync::atomic::Ordering::Relaxed) {
+                break;
+            }
+            let msg = events_recv(client_id, &mut events_sub).await;
+            match msg {
+                Ok(ClientEvent::TaskSuccess(tid, v)) => {
+                    let mut tasks = event_listen_task_results.lock().await;
+                    tasks.insert(tid, Ok(v));
+                }
+                Ok(ClientEvent::TaskError(tid, e)) => {
+                    let mut tasks = event_listen_task_results.lock().await;
+                    tasks.insert(tid, Err(anyhow!("Task error: {:?}", e)));
+                }
+                Ok(_) => {}
+                Err(e) => {
+                    panic!("Error in event recv: {}", e);
+                }
+            }
+        }
+        let seconds_since_start = start_time.elapsed().as_secs();
+        if seconds_since_start % 5 == 0 {
+            let tasks = event_listen_task_results.lock().await;
+            info!(
+                "Event listener running for {} seconds with {} tasks",
+                seconds_since_start,
+                tasks.len()
+            );
+        }
+    });
 }

crates/testing/load-tools/src/tx-list-append.rs

@@ -20,7 +20,9 @@
 
 mod setup;
 
-use crate::setup::{broadcast_handle, create_user_session, initialization_session};
+use crate::setup::{
+    broadcast_handle, create_user_session, initialization_session, listen_responses,
+};
 use clap::Parser;
 use clap_derive::Parser;
 use edn_format::{Keyword, Value};
@@ -423,42 +425,13 @@ async fn list_append_workload(
     .await?;
 
     let task_results = Arc::new(Mutex::new(HashMap::new()));
-
-    let event_listen_task_results = task_results.clone();
-    let ks = kill_switch.clone();
-    tokio::spawn(async move {
-        let start_time = Instant::now();
-        info!("Waiting for events...");
-        loop {
-            if ks.load(std::sync::atomic::Ordering::Relaxed) {
-                break;
-            }
-            let msg = events_recv(client_id, &mut events_sub).await;
-            match msg {
-                Ok(ClientEvent::TaskSuccess(tid, v)) => {
-                    let mut tasks = event_listen_task_results.lock().await;
-                    tasks.insert(tid, Ok(v));
-                }
-                Ok(ClientEvent::TaskError(tid, e)) => {
-                    let mut tasks = event_listen_task_results.lock().await;
-                    tasks.insert(tid, Err(anyhow!("Task error: {:?}", e)));
-                }
-                Ok(_) => {}
-                Err(e) => {
-                    panic!("Error in event recv: {}", e);
-                }
-            }
-        }
-        let seconds_since_start = start_time.elapsed().as_secs();
-        if seconds_since_start % 5 == 0 {
-            let tasks = event_listen_task_results.lock().await;
-            info!(
-                "Event listener running for {} seconds with {} tasks",
-                seconds_since_start,
-                tasks.len()
-            );
-        }
-    });
+    listen_responses(
+        client_id,
+        events_sub,
+        kill_switch.clone(),
+        task_results.clone(),
+    )
+    .await;
 
     info!("Starting {} workloads", args.num_concurrent_workloads);
     let mut workload_futures = FuturesUnordered::new();