qualia_core_db/services/swarm/
executor.rs

1//! Job executors. The [`JobExecutor`] trait is what a worker cell — local or a remote
2//! peer — implements. [`LocalKernelExecutor`] is the **real** one: it runs each job's
3//! kernel through the actual engine (the dynamic GEMM and the real KGE trainer), so a
4//! dispatched job produces genuine computed work — never a stub.
5//!
6//! In the distributed setting the executor is *untrusted* (it could be a stranger's
7//! solar node). That is exactly why dispatch always follows execution with independent
8//! verification ([`super::verify`]) before any payment.
9
10use super::job::{JobInput, JobResult};
11use super::SwarmError;
12use crate::solvers::learning::kg_embedding::{train, EmbeddingTable};
13use crate::solvers::linear_algebra::gemm::matmul;
14
15/// A node that can execute swarm jobs. Object-safe so it can be boxed and swapped
16/// (a local cell, a remote peer proxy, a test double).
17pub trait JobExecutor {
18    fn execute(&self, input: &JobInput) -> Result<JobResult, SwarmError>;
19}
20
21/// Executes jobs on the local CPU through the real engine kernels.
22pub struct LocalKernelExecutor;
23
24impl JobExecutor for LocalKernelExecutor {
25    fn execute(&self, input: &JobInput) -> Result<JobResult, SwarmError> {
26        if !input.is_well_formed() {
27            return Err(SwarmError::InvalidJob);
28        }
29        match input {
30            JobInput::DenseLinearProduct { m, k, n, a, b } => {
31                let mut c = vec![0.0; m * n];
32                matmul(*m, *k, *n, a, b, &mut c).map_err(|_| SwarmError::KernelFailed)?;
33                Ok(JobResult::DenseLinearProduct { c })
34            }
35            JobInput::EmbeddingArtifact {
36                triples,
37                n_entities,
38                n_relations,
39                cfg,
40                ..
41            } => {
42                let table: EmbeddingTable = train(triples, *n_entities, *n_relations, *cfg)
43                    .map_err(|_| SwarmError::KernelFailed)?;
44                Ok(JobResult::EmbeddingArtifact { table })
45            }
46        }
47    }
48}
49
50#[cfg(test)]
51mod tests {
52    use super::*;
53    use crate::solvers::learning::kg_embedding::{ScoreModel, TrainConfig};
54
55    #[test]
56    fn local_executor_computes_a_real_product() {
57        let input = JobInput::DenseLinearProduct {
58            m: 2,
59            k: 2,
60            n: 2,
61            a: vec![1.0, 2.0, 3.0, 4.0],
62            b: vec![5.0, 6.0, 7.0, 8.0],
63        };
64        let r = LocalKernelExecutor.execute(&input).unwrap();
65        match r {
66            // [[1,2],[3,4]]·[[5,6],[7,8]] = [[19,22],[43,50]]
67            JobResult::DenseLinearProduct { c } => {
68                assert_eq!(c, vec![19.0, 22.0, 43.0, 50.0]);
69            }
70            _ => panic!("wrong result kind"),
71        }
72    }
73
74    #[test]
75    fn local_executor_trains_a_real_artifact() {
76        let input = JobInput::EmbeddingArtifact {
77            triples: vec![(0, 0, 1), (2, 0, 3)],
78            n_entities: 4,
79            n_relations: 1,
80            cfg: TrainConfig {
81                model: ScoreModel::TransE { p: 2 },
82                rank: 8,
83                epochs: 200,
84                lr: 0.05,
85                margin: 1.0,
86                reg: 0.0,
87                neg_per_pos: 4,
88                seed: 7,
89            },
90            check: vec![(0, 0, 1)],
91        };
92        let r = LocalKernelExecutor.execute(&input).unwrap();
93        match r {
94            JobResult::EmbeddingArtifact { table } => {
95                // The trained table scores the true tail above a wrong one.
96                assert!(table.score(0, 0, 1).unwrap() > table.score(0, 0, 3).unwrap());
97            }
98            _ => panic!("wrong result kind"),
99        }
100    }
101
102    #[test]
103    fn malformed_job_fails_closed() {
104        let bad = JobInput::DenseLinearProduct {
105            m: 2,
106            k: 2,
107            n: 2,
108            a: vec![1.0],
109            b: vec![1.0; 4],
110        };
111        assert_eq!(
112            LocalKernelExecutor.execute(&bad).unwrap_err(),
113            SwarmError::InvalidJob
114        );
115    }
116}
qualia_core_db/services/swarm/executor.rs

qualia_core_db/services/swarm/
executor.rs
