Rip Rayon out of reducer execution and query evaluation

crates/core/src/host/wasm_common/module_host_actor.rs

@@ -462,9 +462,9 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         )
         .entered();
 
-        // run the call_reducer call in rayon. it's important that we don't acquire a lock inside a rayon task,
-        // as that can lead to deadlock.
-        let (mut tx, result) = rayon::scope(|_| tx_slot.set(tx, || self.instance.call_reducer(op, budget)));
+        // FOR BENCHMARKING: Just run the reducer on whatever thread we're already on,
+        // instead of bouncing to a Rayon thread.
+        let (mut tx, result) = tx_slot.set(tx, || self.instance.call_reducer(op, budget));
 
         let ExecuteResult {
             energy,

crates/core/src/subscription/mod.rs

@@ -2,7 +2,6 @@ use std::sync::Arc;
 
 use anyhow::Result;
 use module_subscription_manager::Plan;
-use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use spacetimedb_client_api_messages::websocket::{
     ByteListLen, Compression, DatabaseUpdate, QueryUpdate, TableUpdate, WebsocketFormat,
 };
@@ -112,8 +111,9 @@ where
     Tx: Datastore + DeltaStore + Sync,
     F: WebsocketFormat,
 {
+    // FOR TESTING: Just evaluate sequentially.
     plans
-        .par_iter()
+        .iter()
         .map(|plan| (plan, plan.subscribed_table_id(), plan.subscribed_table_name()))
         .map(|(plan, table_id, table_name)| {
             plan.physical_plan()

crates/core/src/subscription/module_subscription_manager.rs

@@ -13,7 +13,6 @@ use crate::subscription::delta::eval_delta;
 use crate::subscription::record_exec_metrics;
 use hashbrown::hash_map::OccupiedError;
 use hashbrown::{HashMap, HashSet};
-use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use spacetimedb_client_api_messages::websocket::{
     BsatnFormat, CompressableQueryUpdate, Compression, FormatSwitch, JsonFormat, QueryId, QueryUpdate, WebsocketFormat,
 };
@@ -463,8 +462,8 @@ impl SubscriptionManager {
 
         let tables = &event.status.database_update().unwrap().tables;
 
-        // Put the main work on a rayon compute thread.
-        rayon::scope(|_| {
+        // FOR TESTING: Just do all this work on whatever the calling thread is.
+        {
             let span = tracing::info_span!("eval_incr").entered();
             let (updates, errs, metrics) = tables
                 .iter()
@@ -473,15 +472,16 @@ impl SubscriptionManager {
                 .filter_map(|table_id| self.tables.get(table_id))
                 .flatten()
                 .collect::<HashSet<_>>()
-                .par_iter()
+            // FOR TESTING: Sequential execution, rather than parallel.
+                .iter()
                 .filter_map(|&hash| {
                     self.queries
                         .get(hash)
                         .map(|state| (hash, &state.query, ExecutionMetrics::default()))
                 })
-                // If N clients are subscribed to a query,
-                // we copy the DatabaseTableUpdate N times,
-                // which involves cloning BSATN (binary) or product values (json).
+            // If N clients are subscribed to a query,
+            // we copy the DatabaseTableUpdate N times,
+            // which involves cloning BSATN (binary) or product values (json).
                 .map(|(hash, plan, mut metrics)| {
                     let table_id = plan.subscribed_table_id();
                     let table_name: Box<str> = plan.subscribed_table_name().into();
@@ -561,7 +561,7 @@ impl SubscriptionManager {
                     (updates, metrics)
                 })
                 .fold(
-                    || (vec![], vec![], ExecutionMetrics::default()),
+                    (vec![], vec![], ExecutionMetrics::default()),
                     |(mut rows, mut errs, mut agg_metrics), (result, metrics)| {
                         match result {
                             Ok(x) => {
@@ -575,13 +575,14 @@ impl SubscriptionManager {
                         (rows, errs, agg_metrics)
                     },
                 )
-                .reduce_with(|(mut acc_rows, mut acc_errs, mut acc_metrics), (rows, errs, metrics)| {
-                    acc_rows.extend(rows);
-                    acc_errs.extend(errs);
-                    acc_metrics.merge(metrics);
-                    (acc_rows, acc_errs, acc_metrics)
-                })
-                .unwrap_or_default();
+            // .reduce_with(|(mut acc_rows, mut acc_errs, mut acc_metrics), (rows, errs, metrics)| {
+            //     acc_rows.extend(rows);
+            //     acc_errs.extend(errs);
+            //     acc_metrics.merge(metrics);
+            //     (acc_rows, acc_errs, acc_metrics)
+            // })
+            // .unwrap_or_default()
+                ;
 
             record_exec_metrics(&WorkloadType::Update, database_identity, metrics);
 
@@ -685,7 +686,7 @@ impl SubscriptionManager {
                     );
                 }
             }
-        })
+        }
     }
 }
 

crates/core/src/subscription/subscription.rs

@@ -32,7 +32,6 @@ use crate::sql::ast::SchemaViewer;
 use crate::vm::{build_query, TxMode};
 use anyhow::Context;
 use itertools::Either;
-use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use spacetimedb_client_api_messages::websocket::{Compression, WebsocketFormat};
 use spacetimedb_data_structures::map::HashSet;
 use spacetimedb_lib::db::auth::{StAccess, StTableType};
@@ -519,11 +518,11 @@ impl ExecutionSet {
         slow_query_threshold: Option<Duration>,
         compression: Compression,
     ) -> ws::DatabaseUpdate<F> {
-        // evaluate each of the execution units in this ExecutionSet in parallel
+        // FOR TESTING: Just do sequential execution.
         let tables = self
             .exec_units
             // if you need eval to run single-threaded for debugging, change this to .iter()
-            .par_iter()
+            .iter()
             .filter_map(|unit| unit.eval(db, tx, &unit.sql, slow_query_threshold, compression))
             .collect();
         ws::DatabaseUpdate { tables }