Thread jobs.rs changes through all of core crate.

Author: gefjon

crates/core/src/host/host_controller.rs

@@ -14,7 +14,7 @@ use crate::replica_context::ReplicaContext;
 use crate::subscription::module_subscription_actor::ModuleSubscriptions;
 use crate::subscription::module_subscription_manager::{spawn_send_worker, SubscriptionManager};
 use crate::util::asyncify;
-use crate::util::jobs::{JobCore, JobCores};
+use crate::util::jobs::{JobCores, SingleCoreExecutor};
 use crate::worker_metrics::WORKER_METRICS;
 use anyhow::{anyhow, ensure, Context};
 use async_trait::async_trait;
@@ -698,7 +698,7 @@ async fn make_module_host(
     program: Program,
     energy_monitor: Arc<dyn EnergyMonitor>,
     unregister: impl Fn() + Send + Sync + 'static,
-    core: JobCore,
+    executor: SingleCoreExecutor,
 ) -> anyhow::Result<(Program, ModuleHost)> {
     // `make_actor` is blocking, as it needs to compile the wasm to native code,
     // which may be computationally expensive - sometimes up to 1s for a large module.
@@ -718,12 +718,12 @@ async fn make_module_host(
             HostType::Wasm => {
                 let actor = runtimes.wasmtime.make_actor(mcc)?;
                 trace!("wasmtime::make_actor blocked for {:?}", start.elapsed());
-                ModuleHost::new(actor, unregister, core)
+                ModuleHost::new(actor, unregister, executor)
             }
             HostType::Js => {
                 let actor = runtimes.v8.make_actor(mcc)?;
                 trace!("v8::make_actor blocked for {:?}", start.elapsed());
-                ModuleHost::new(actor, unregister, core)
+                ModuleHost::new(actor, unregister, executor)
             }
         };
         Ok((program, module_host))
@@ -756,7 +756,7 @@ async fn launch_module(
     energy_monitor: Arc<dyn EnergyMonitor>,
     replica_dir: ReplicaDir,
     runtimes: Arc<HostRuntimes>,
-    core: JobCore,
+    executor: SingleCoreExecutor,
 ) -> anyhow::Result<(Program, LaunchedModule)> {
     let db_identity = database.database_identity;
     let host_type = database.host_type;
@@ -773,7 +773,7 @@ async fn launch_module(
         program,
         energy_monitor.clone(),
         on_panic,
-        core,
+        executor,
     )
     .await?;
 
@@ -991,7 +991,7 @@ impl Host {
         page_pool: PagePool,
         database: Database,
         program: Program,
-        core: JobCore,
+        executor: SingleCoreExecutor,
     ) -> anyhow::Result<Arc<ModuleInfo>> {
         // Even in-memory databases acquire a lockfile.
         // Grab a tempdir to put that lockfile in.
@@ -1024,7 +1024,7 @@ impl Host {
             Arc::new(NullEnergyMonitor),
             phony_replica_dir,
             runtimes.clone(),
-            core,
+            executor,
         )
         .await?;
 
@@ -1057,7 +1057,7 @@ impl Host {
         policy: MigrationPolicy,
         energy_monitor: Arc<dyn EnergyMonitor>,
         on_panic: impl Fn() + Send + Sync + 'static,
-        core: JobCore,
+        executor: SingleCoreExecutor,
     ) -> anyhow::Result<UpdateDatabaseResult> {
         let replica_ctx = &self.replica_ctx;
         let (scheduler, scheduler_starter) = Scheduler::open(self.replica_ctx.relational_db.clone());
@@ -1070,7 +1070,7 @@ impl Host {
             program,
             energy_monitor,
             on_panic,
-            core,
+            executor,
         )
         .await?;
 
@@ -1187,7 +1187,7 @@ pub async fn extract_schema(program_bytes: Box<[u8]>, host_type: HostType) -> an
 
     let runtimes = HostRuntimes::new(None);
     let page_pool = PagePool::new(None);
-    let core = JobCore::default();
+    let core = SingleCoreExecutor::in_current_tokio_runtime();
     let module_info = Host::try_init_in_memory_to_check(&runtimes, page_pool, database, program, core).await?;
     // this should always succeed, but sometimes it doesn't
     let module_def = match Arc::try_unwrap(module_info) {

crates/core/src/host/module_common.rs

@@ -3,10 +3,7 @@
 
 use crate::{
     energy::EnergyMonitor,
-    host::{
-        module_host::{DynModule, ModuleInfo},
-        Scheduler,
-    },
+    host::{module_host::ModuleInfo, Scheduler},
     module_host_context::ModuleCreationContext,
     replica_context::ReplicaContext,
 };
@@ -79,12 +76,12 @@ impl ModuleCommon {
     }
 }
 
-impl DynModule for ModuleCommon {
-    fn replica_ctx(&self) -> &Arc<ReplicaContext> {
+impl ModuleCommon {
+    pub fn replica_ctx(&self) -> &Arc<ReplicaContext> {
         &self.replica_context
     }
 
-    fn scheduler(&self) -> &Scheduler {
+    pub fn scheduler(&self) -> &Scheduler {
         &self.scheduler
     }
 }