1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
//! How a signal moves: the gate, the firing, and the pathway its timing selects
use std::sync::atomic::Ordering;
use std::time::Instant;
use tracing::{debug, warn};
use super::{Cell, Fabric, MAX_HOPS, Outcome, Pathway};
use crate::nerve::latency::LatencyEstimate;
use crate::nerve::neuron::{Cx, Firing, NeuronKind, Quiet};
use crate::nerve::signal::{NeuronId, PathwayId, Signal};
impl Fabric {
/// Hand a signal to a receptor: the whole synchronous chain runs before this returns.
/// Receptors call this through their [`Afferent`](crate::nerve::receptor::Afferent)
/// handle; tests and in-process sources call it directly.
pub fn stimulate(&self, receptor: NeuronId, signal: Signal) -> Outcome {
self.stimulate_announced(receptor, signal, None)
}
/// Announce a signal that will take `pathway` once it is ready (its request's body is
/// still being received or decoded): a worker collecting a batch there waits for it.
/// Hold the guard while preparing the signal, then hand both to
/// [`Fabric::stimulate_announced`]. `None` for a reflex pathway, which has no queue.
pub(crate) fn announce(&self, pathway: PathwayId) -> Option<Announced<'_>> {
let queue = self.0.pathways.get(pathway.index())?.queue.as_ref()?;
Some(Announced {
pathway,
arrival: queue.arrival(),
})
}
/// [`Fabric::stimulate`] for a signal announced with [`Fabric::announce`]: if it takes
/// the announced pathway, it is queued as the arrival a worker may be waiting for
pub(crate) fn stimulate_announced(
&self,
receptor: NeuronId,
mut signal: Signal,
announced: Option<Announced<'_>>,
) -> Outcome {
signal.origin = receptor;
self.0.received.fetch_add(1, Ordering::Relaxed);
let mut announced = announced;
self.drive(receptor, signal, 0, &mut announced, Instant::now())
}
/// [`Fabric::stimulate`] by the receptor's name
pub fn stimulate_named(&self, receptor: &str, signal: Signal) -> Option<Outcome> {
Some(self.stimulate(self.neuron(receptor)?, signal))
}
/// Run `id`'s gate and `stimulate`, then propagate what it fires; `now` is when the
/// signal arrived, which the caller has read already
fn drive(
&self,
id: NeuronId,
mut signal: Signal,
depth: usize,
announced: &mut Option<Announced<'_>>,
now: Instant,
) -> Outcome {
let cell = &self.0.cells[id.index()];
let nanos = self.nanos(now);
cell.counters.received.fetch_add(1, Ordering::Relaxed);
if let Err(reason) = cell.gate.admit(&mut signal, nanos) {
cell.counters.quiet(reason);
return Outcome {
quiet: Some((id, reason)),
hops: depth,
..Outcome::default()
};
}
signal.trace.push(id);
let mut acted = 0;
match cell.kind {
NeuronKind::Brain => {
signal.trace.mark_brain();
self.0.reached_brain.fetch_add(1, Ordering::Relaxed);
}
NeuronKind::Effector => {
acted = 1;
if !signal.trace.was_queued()
&& let Some(arc) = cell.arcs.get(signal.origin.index())
{
arc.get_or_init(LatencyEstimate::new)
.observe(now.saturating_duration_since(signal.at));
}
}
NeuronKind::Receptor | NeuronKind::Reflex => {}
}
let cx = Cx {
now,
id,
fabric: self,
};
let firing = cell.neuron.stimulate(signal, &cx);
if matches!(firing, Firing::Quiet) {
cell.gate.stayed_quiet(nanos);
}
let mut outcome = self.after_firing(id, firing, depth, announced);
outcome.acted += acted;
outcome
}
/// Propagate what `id` fired
pub(super) fn after_firing(
&self,
id: NeuronId,
firing: Firing,
depth: usize,
announced: &mut Option<Announced<'_>>,
) -> Outcome {
let cell = &self.0.cells[id.index()];
let emitted = match firing {
Firing::Quiet => {
cell.counters.quiet(Quiet::NotInterested);
return Outcome {
quiet: Some((id, Quiet::NotInterested)),
hops: depth + 1,
..Outcome::default()
};
}
Firing::Fire(signals) => signals,
Firing::Compute(work) => {
cell.counters.computed.fetch_add(1, Ordering::Relaxed);
work()
}
};
cell.counters.fired.fetch_add(1, Ordering::Relaxed);
let mut outcome = Outcome {
hops: depth + 1,
..Outcome::default()
};
for signal in emitted {
outcome.merge(self.route(id, signal, depth + 1, announced));
}
outcome
}
/// Send a signal `from` fired along the pathway its timing selects
fn route(
&self,
from: NeuronId,
mut signal: Signal,
depth: usize,
announced: &mut Option<Announced<'_>>,
) -> Outcome {
// The depth restarts after a deliberative queue; the trace keeps the whole count, so a
// loop through a queue is cut too
if depth >= MAX_HOPS || signal.trace.len() >= MAX_HOPS {
self.0.loops.fetch_add(1, Ordering::Relaxed);
warn!(
"nerve: signal {} dropped after {MAX_HOPS} hops (a loop?)",
signal.id.0
);
return Outcome::default();
}
let now = Instant::now();
let cell = &self.0.cells[from.index()];
let selected = match signal.payload.as_command() {
// A command names its effector; the pathway to it is the one (declared or not)
Some(command) => match self.0.by_name.get(&command.effector) {
Some(&target) if self.0.cells[target.index()].kind == NeuronKind::Effector => {
match cell
.out
.iter()
.map(|&p| &self.0.pathways[p.index()])
.find(|p| p.to == target)
{
Some(pathway) => Some(pathway),
None => {
self.0.commands_direct.fetch_add(1, Ordering::Relaxed);
return self.drive(target, signal, depth, announced, now);
}
}
}
_ => {
self.0.unrouted.fetch_add(1, Ordering::Relaxed);
debug!(
"nerve: no effector named '{}' for a command from {}",
command.effector,
cell.neuron.name()
);
return Outcome {
unrouted: 1,
..Outcome::default()
};
}
},
None => self.select(cell, &signal, now),
};
let Some(pathway) = selected else {
if cell.out.is_empty() {
// A terminal neuron: nothing to do with what it fired
return Outcome::default();
}
self.0.no_eligible.fetch_add(1, Ordering::Relaxed);
return Outcome {
no_pathway: 1,
..Outcome::default()
};
};
pathway.signals.fetch_add(1, Ordering::Relaxed);
match &pathway.queue {
None => {
let outcome = self.drive(pathway.to, signal, depth, announced, now);
pathway.latency.observe(now.elapsed());
outcome
}
Some(queue) => {
self.prepare_for(pathway, &signal, now);
signal.trace.mark_queued();
// The arrival a worker may be waiting for, if this is the announced pathway
let arrival = match announced.take() {
Some(a) if a.pathway == pathway.id => Some(a.arrival),
other => {
*announced = other;
None
}
};
match queue.submit(arrival, signal) {
Ok(()) => Outcome {
queued: 1,
..Outcome::default()
},
// Full: dropped rather than queued without bound
Err(_) => {
pathway.overflow.fetch_add(1, Ordering::Relaxed);
Outcome {
dropped: 1,
..Outcome::default()
}
}
}
}
}
}
/// The short-mode / long-mode rule: among the pathways leaving `cell` whose p99 fits in
/// the signal's remaining deadline, the heaviest weight
fn select(&self, cell: &Cell, signal: &Signal, now: Instant) -> Option<&Pathway> {
let remaining = signal.urgency.remaining(now);
let mut best: Option<&Pathway> = None;
for &id in &cell.out {
let pathway = &self.0.pathways[id.index()];
if !pathway.fits(remaining) {
pathway.ineligible.fetch_add(1, Ordering::Relaxed);
continue;
}
if best.is_none_or(|b| pathway.weight > b.weight) {
best = Some(pathway);
}
}
best
}
/// Motor preparation: a signal with time to spare that takes a deliberative pathway
/// warms the neurons it will probably reach, at most once a second each
fn prepare_for(&self, pathway: &Pathway, signal: &Signal, now: Instant) {
let far = match (signal.urgency.remaining(now), pathway.latency.p99()) {
(None, _) | (_, None) => true,
(Some(remaining), Some(p99)) => remaining > 2 * p99,
};
if !far {
return;
}
let nanos = self.nanos(now);
let to = &self.0.cells[pathway.to.index()];
let downstream = to
.out
.iter()
.map(|&p| self.0.pathways[p.index()].to)
.chain(std::iter::once(pathway.to));
for id in downstream {
let cell = &self.0.cells[id.index()];
let last = cell.last_prepared.load(Ordering::Relaxed);
if nanos.saturating_sub(last) < 1_000_000_000 && last != 0 {
continue;
}
cell.last_prepared.store(nanos.max(1), Ordering::Relaxed);
let cx = Cx {
now,
id,
fabric: self,
};
cell.neuron.prepare(&cx);
}
}
}
/// A signal announced on a deliberative pathway, see [`Fabric::announce`]
pub(crate) struct Announced<'a> {
pathway: PathwayId,
arrival: amate::Arrival<'a, Signal>,
}