furnace/nerve/fabric/
builder.rs

1//! Building a fabric: neurons, receptors, effectors, and the pathways between them
2
3use std::collections::HashMap;
4use std::sync::atomic::{AtomicBool, AtomicU64};
5use std::sync::{Arc, Mutex, OnceLock, Weak};
6use std::time::Instant;
7
8use super::pathway::{Pathway, PathwayKind, Summation};
9use super::summing::Summing;
10use super::{Cell, Counters, Fabric, Inner};
11use crate::nerve::effector::Effector;
12use crate::nerve::latency::LatencyEstimate;
13use crate::nerve::neuron::{Gate, Mechanisms, Neuron, NeuronKind};
14use crate::nerve::receptor::{Receptor, Stop};
15use crate::nerve::signal::{NeuronId, PathwayId};
16use amate::Engine;
17
18/// Why a part of the nervous system cannot be built
19#[derive(Debug)]
20#[non_exhaustive]
21pub enum BuildError {
22    /// Two neurons have the same name; effectors are addressed by name
23    DuplicateName(String),
24    /// A pathway names a neuron that was not added to this builder
25    UnknownNeuron {
26        /// The pathway's source
27        from: NeuronId,
28        /// The pathway's target
29        to: NeuronId,
30    },
31    /// A URL the nervous system's HTTP client cannot call: not `http://host[:port]/path`
32    /// (HTTPS among them)
33    Url {
34        /// The effector or receptor
35        name: String,
36        /// What is wrong with the URL
37        reason: String,
38    },
39    /// A runtime of the fabric could not be started
40    Runtime(&'static str, std::io::Error),
41}
42
43impl std::fmt::Display for BuildError {
44    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        match self {
46            Self::DuplicateName(name) => write!(f, "two neurons are named '{name}'"),
47            Self::UnknownNeuron { from, to } => {
48                write!(
49                    f,
50                    "pathway {from} -> {to} names a neuron that was not added"
51                )
52            }
53            Self::Url { name, reason } => write!(f, "{name}: {reason}"),
54            Self::Runtime(which, e) => write!(f, "the {which} runtime: {e}"),
55        }
56    }
57}
58
59impl std::error::Error for BuildError {
60    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
61        match self {
62            Self::Runtime(_, e) => Some(e),
63            _ => None,
64        }
65    }
66}
67
68/// Builds a [`Fabric`]: add neurons, connect them, build
69pub struct FabricBuilder {
70    neurons: Vec<(Arc<dyn Neuron>, Mechanisms)>,
71    receptors: Vec<(NeuronId, Arc<dyn Receptor>)>,
72    pathways: Vec<(NeuronId, NeuronId, PathwayKind, f32, Option<Summation>)>,
73    efferent_threads: usize,
74    efferent: Option<tokio::runtime::Handle>,
75}
76
77impl Default for FabricBuilder {
78    fn default() -> Self {
79        Self::new()
80    }
81}
82
83impl FabricBuilder {
84    /// An empty builder
85    pub fn new() -> Self {
86        Self {
87            neurons: Vec::new(),
88            receptors: Vec::new(),
89            pathways: Vec::new(),
90            efferent_threads: 2,
91            efferent: None,
92        }
93    }
94
95    /// Threads of the motor side's runtime (2 by default)
96    pub fn efferent_threads(mut self, threads: usize) -> Self {
97        self.efferent_threads = threads.max(1);
98        self
99    }
100
101    /// Run the motor side (effectors' calls, retries, backoff) on a runtime of your own, the
102    /// server's say, instead of one the fabric starts. The runtime must outlive the fabric;
103    /// [`efferent_threads`](Self::efferent_threads) no longer applies.
104    pub fn efferent(mut self, runtime: tokio::runtime::Handle) -> Self {
105        self.efferent = Some(runtime);
106        self
107    }
108
109    /// Add a neuron, with no mechanisms in front of it ([`gate`](Self::gate) adds them).
110    /// Names must be unique, since effectors are addressed by name in commands:
111    /// [`build`](Self::build) refuses two of the same name.
112    pub fn neuron(&mut self, neuron: impl Neuron) -> NeuronId {
113        self.add(Arc::new(neuron))
114    }
115
116    /// [`FabricBuilder::neuron`] for a neuron already shared
117    pub(crate) fn add(&mut self, neuron: Arc<dyn Neuron>) -> NeuronId {
118        let id = NeuronId(u32::try_from(self.neurons.len()).expect("fewer than 4G neurons"));
119        self.neurons.push((neuron, Mechanisms::NONE));
120        id
121    }
122
123    /// Add a receptor: a neuron that is also started with the fabric and may be mounted as a
124    /// route. Habituation belongs here ([`gate`](Self::gate)).
125    pub fn receptor<R: Receptor>(&mut self, receptor: R) -> NeuronId {
126        let receptor: Arc<R> = Arc::new(receptor);
127        let neuron: Arc<dyn Neuron> = receptor.clone();
128        let id = self.add(neuron);
129        self.receptors.push((id, receptor));
130        id
131    }
132
133    /// Add an effector: an [`Act`](crate::nerve::effector::Act) with the default local loop
134    /// (`b.effector(Webhook::new(..)?)`), or an [`Effector`] configured first. Commands
135    /// reach it by its name from any neuron, with or without a pathway to it.
136    pub fn effector(&mut self, effector: impl Into<Effector>) -> NeuronId {
137        self.neuron(effector.into())
138    }
139
140    /// Put `mechanisms` (threshold, habituation, refractory period) in front of neuron `id`
141    ///
142    /// # Panics
143    ///
144    /// For an id this builder did not hand out.
145    pub fn gate(&mut self, id: NeuronId, mechanisms: Mechanisms) -> &mut Self {
146        self.neurons[id.index()].1 = mechanisms;
147        self
148    }
149
150    /// The id the next pathway added will have
151    pub(crate) fn next_pathway(&self) -> PathwayId {
152        PathwayId(u32::try_from(self.pathways.len()).expect("fewer than 4G pathways"))
153    }
154
155    /// A reflex pathway: `to` is called directly when `from` fires. Weight 1 unless
156    /// [`Link::weight`] says otherwise.
157    pub fn reflex(&mut self, from: NeuronId, to: NeuronId) -> Link<'_> {
158        self.link(from, to, PathwayKind::Reflex, REFLEX_WEIGHT, None)
159    }
160
161    /// A deliberative pathway: signals from `from` queue to a worker that sums them into
162    /// firings of `to`. Weight 2, so it wins over a reflex pathway whenever the signal's
163    /// deadline allows it, and the default [`Summation`]; [`Link`] changes either.
164    pub fn deliberative(&mut self, from: NeuronId, to: NeuronId) -> Link<'_> {
165        self.link(
166            from,
167            to,
168            PathwayKind::Deliberative,
169            DELIBERATIVE_WEIGHT,
170            Some(Summation::default()),
171        )
172    }
173
174    fn link(
175        &mut self,
176        from: NeuronId,
177        to: NeuronId,
178        kind: PathwayKind,
179        weight: f32,
180        summation: Option<Summation>,
181    ) -> Link<'_> {
182        self.pathways.push((from, to, kind, weight, summation));
183        Link {
184            pathway: self.pathways.last_mut().expect("just pushed"),
185        }
186    }
187
188    /// Wire the fabric and start its runtimes and its deliberative workers. An error for two
189    /// neurons of the same name, a pathway to a neuron of another builder, or a runtime that
190    /// cannot be started.
191    pub fn build(self) -> Result<Fabric, BuildError> {
192        let count = self.neurons.len();
193        let mut names = std::collections::HashSet::new();
194        for (neuron, _) in &self.neurons {
195            if !names.insert(neuron.name()) {
196                return Err(BuildError::DuplicateName(neuron.name().to_string()));
197            }
198        }
199        drop(names);
200        for &(from, to, ..) in &self.pathways {
201            if from.index() >= count || to.index() >= count {
202                return Err(BuildError::UnknownNeuron { from, to });
203            }
204        }
205        let (efferent, efferent_handle) = match self.efferent {
206            Some(handle) => (None, handle),
207            None => {
208                let runtime = tokio::runtime::Builder::new_multi_thread()
209                    .worker_threads(self.efferent_threads)
210                    .thread_name("furnace-efferent")
211                    .enable_all()
212                    .build()
213                    .map_err(|e| BuildError::Runtime("efferent", e))?;
214                let handle = runtime.handle().clone();
215                (Some(runtime), handle)
216            }
217        };
218        let afferent = tokio::runtime::Builder::new_multi_thread()
219            .worker_threads(1)
220            .thread_name("furnace-afferent")
221            .enable_all()
222            .build()
223            .map_err(|e| BuildError::Runtime("afferent", e))?;
224        let mut by_name = HashMap::new();
225        let mut cells: Vec<Cell> = self
226            .neurons
227            .into_iter()
228            .enumerate()
229            .map(|(i, (neuron, mechanisms))| {
230                by_name.insert(neuron.name().to_string(), NeuronId(i as u32));
231                let kind = neuron.kind();
232                Cell {
233                    arcs: if kind == NeuronKind::Effector {
234                        (0..count).map(|_| OnceLock::new()).collect()
235                    } else {
236                        Vec::new()
237                    },
238                    neuron,
239                    kind,
240                    gate: Gate::new(mechanisms),
241                    counters: Counters::default(),
242                    out: Vec::new(),
243                    last_prepared: AtomicU64::new(0),
244                }
245            })
246            .collect();
247        let inner = Arc::new_cyclic(|weak: &Weak<Inner>| {
248            let pathways: Vec<Pathway> = self
249                .pathways
250                .into_iter()
251                .enumerate()
252                .map(|(i, (from, to, kind, weight, summation))| {
253                    let id = PathwayId(i as u32);
254                    cells[from.index()].out.push(id);
255                    let queue = summation.map(|s| {
256                        Engine::start(
257                            Summing {
258                                fabric: weak.clone(),
259                                pathway: id,
260                            },
261                            amate::Config {
262                                max_batch: s.max_batch,
263                                workers: s.workers,
264                                capacity: Some(s.capacity),
265                                waits_unannounced: !s.announced_only,
266                            },
267                            move |_| format!("furnace-nerve-{i}"),
268                        )
269                    });
270                    Pathway {
271                        id,
272                        from,
273                        to,
274                        kind,
275                        weight,
276                        latency: LatencyEstimate::new(),
277                        signals: AtomicU64::new(0),
278                        ineligible: AtomicU64::new(0),
279                        overflow: AtomicU64::new(0),
280                        summation,
281                        queue,
282                    }
283                })
284                .collect();
285            Inner {
286                epoch: Instant::now(),
287                cells,
288                pathways,
289                by_name,
290                receptors: self.receptors,
291                started: AtomicBool::new(false),
292                received: AtomicU64::new(0),
293                reached_brain: AtomicU64::new(0),
294                unrouted: AtomicU64::new(0),
295                no_eligible: AtomicU64::new(0),
296                loops: AtomicU64::new(0),
297                commands_direct: AtomicU64::new(0),
298                stop: Arc::new(Stop::default()),
299                efferent_handle,
300                efferent: Mutex::new(efferent),
301                afferent_handle: afferent.handle().clone(),
302                afferent: Mutex::new(Some(afferent)),
303            }
304        });
305        Ok(Fabric(inner))
306    }
307}
308
309/// The weight of a reflex pathway unless [`Link::weight`] changes it
310const REFLEX_WEIGHT: f32 = 1.0;
311/// The weight of a deliberative pathway unless [`Link::weight`] changes it: above a
312/// reflex's, so the brain is chosen whenever there is time for it
313const DELIBERATIVE_WEIGHT: f32 = 2.0;
314
315/// A pathway just added, to change its weight or summation
316pub struct Link<'a> {
317    pathway: &'a mut (NeuronId, NeuronId, PathwayKind, f32, Option<Summation>),
318}
319
320impl Link<'_> {
321    /// Priority among the eligible pathways from the same neuron: the heaviest wins
322    pub fn weight(self, weight: f32) -> Self {
323        self.pathway.3 = weight;
324        self
325    }
326
327    /// How a deliberative pathway sums its queue into firings. On a reflex pathway it makes
328    /// the pathway deliberative.
329    pub fn summation(self, summation: Summation) -> Self {
330        self.pathway.2 = PathwayKind::Deliberative;
331        self.pathway.4 = Some(summation);
332        self
333    }
334}
furnace/nerve/fabric/builder.rs

furnace/nerve/fabric/
builder.rs
