aboutsummaryrefslogtreecommitdiffstats
path: root/src/api
diff options
context:
space:
mode:
Diffstat (limited to 'src/api')
-rw-r--r--src/api/dashboard/dashboard.html154
-rw-r--r--src/api/dashboard/mod.rs80
2 files changed, 129 insertions, 105 deletions
diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index 05c233c..b7b45aa 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -2897,12 +2897,11 @@ function unpin(id) {
graph.addEventListener("mousemove", event => { if (event.target === graph) hover(null); });
graph.addEventListener("mouseleave", () => hover(null));
-// --- Messages on the map: this node's traffic with each peer (`/peers`, `traffic`),
-// as it grew since the last update, and each subscribed node's messages with its peers
-// (`remoteMessagesArrived`), drawn as dots running along the link the way the messages
-// went. Counts come with each update (every 2 s), so the dots are spread
-// over the next 2 s, a few at most per link and way: a picture of the traffic, not a
-// record of it. As dots, all at one speed (`flowSpeed`), so over a longer link they
+// --- Messages on the map: this node's messages with its peers (`ownMessagesArrived`)
+// and each subscribed node's (`remoteMessagesArrived`), as their streams bring them,
+// soon after they went, drawn as dots running along the link the way the messages
+// went, as far apart as they were sent, a few at most per link and way of a batch: a
+// picture of the traffic, not a record of it. As dots, all at one speed (`flowSpeed`), so over a longer link they
// arrive later; as flashes, each crosses its link as fast as the message did (half the
// link's round trip) and leaves an afterglow that draws back into where it went
// (`FLASH_GLOW_MS`). Dots are kept above everything else on the map (`flows`, after
@@ -2980,9 +2979,6 @@ function oneWayMs(from, to) {
return FLASH_UNKNOWN_MS;
}
let lastTraffic = null;
-// When the last update's traffic came, by the wall clock (it runs on while the computer
-// sleeps, unlike `performance.now()` on some systems).
-let lastTrafficAt = 0;
const flows = [];
let flowFrame = null;
// Messages and bytes by kind, of a peer's `traffic.sent` or `.received`.
@@ -3058,17 +3054,12 @@ function dotsFor(kinds, total) {
}
return dots;
}
-// The messages exchanged with each peer since the last update, on the map (`flows`, if
-// shown).
+// This node's traffic with each peer, as the update brings it: its rates, and what its
+// peers should have received (`deliveredHere`). The map runs messages from the message
+// logs instead, as they come (`runMessages`).
function trafficArrived() {
if (!state) return;
- const self = state.node.id, now = performance.now();
- // Only for an update in its time: not the first after a gap (the computer slept, the
- // stream stalled), whose counts grew over all of it, nor those queued meanwhile and
- // arriving at once with it, each of which would add its dots to the same 2 s.
- const wall = Date.now(), gap = wall - lastTrafficAt;
- lastTrafficAt = wall;
- const dots = dotsNow(gap);
+ const self = state.node.id;
const current = new Map();
for (const [id, peer] of Object.entries(state.peers.peers)) {
if (peer.traffic) {
@@ -3076,53 +3067,76 @@ function trafficArrived() {
received: messages(peer.traffic.received) });
}
}
- // What a subscribed node sent this one runs from its own messages, as it left
- // (`remoteMessagesArrived`), not from what this one counted received.
- if (lastTraffic && dots) addFlows(self, current, lastTraffic,
- (way, id) => way === "received" && logging(id));
if (lastTraffic) deliveredHere(self, current, lastTraffic);
measureRates(current);
lastTraffic = current;
}
-// Dots for `node`'s traffic with each peer, as it grew from `before` to `current` (by
-// peer: `sent`, `received`, each messages and bytes by kind), spread over the next 2 s;
-// but where `skip(way, peer)` (drawn already from another's).
-function addFlows(node, current, before, skip) {
+// Whether messages run on the map now: shown, and seen.
+function flowsNow() {
+ return flowsShown && !calm.matches && !document.hidden &&
+ !document.getElementById("map").classList.contains("collapsed");
+}
+// Messages as a stream brought them, this node's or a subscribed node's (each named by
+// both ends, `from` and `to`), run along their links as they come: started as far
+// apart as they were sent (`at`, by the clock of the node that logged them, so only
+// against each other), `FLOWS_AT_MOST` at most per link and way of a batch, shared out
+// by kind (`dotsFor`) where more came. Of a batch that spans more than
+// `FLOW_SPREAD_MS` (one after a gap: the computer slept, the stream stalled), only
+// those of its last `FLOW_SPREAD_MS`.
+function runMessages(list) {
+ if (!flowsNow() || list.length === 0) return;
const now = performance.now();
- for (const [id, counts] of current) {
- const was = before.get(id);
- if (!was) continue;
- for (const [way, from, to] of [["sent", node, id], ["received", id, node]]) {
- if (skip(way, id)) continue;
- const kinds = {};
- let total = 0;
- for (const [kind, count] of Object.entries(counts[way])) {
- const earlier = was[way][kind] || { messages: 0, bytes: 0 };
- const grown = Math.max(0, count.messages - earlier.messages);
- if (grown > 0) {
- kinds[kind] = { messages: grown, bytes: Math.max(0, count.bytes - earlier.bytes) };
- total += grown;
- }
+ const latest = Math.max(...list.map(message => message.at));
+ const recent = list.filter(message => latest - message.at <= FLOW_SPREAD_MS);
+ const first = Math.min(...recent.map(message => message.at));
+ const links = new Map();
+ for (const message of recent) {
+ const transport = schemeOf(message.address);
+ const key = message.from + " " + message.to + " " + (transport || "");
+ const link = links.get(key) ||
+ { from: message.from, to: message.to, transport, messages: [] };
+ link.messages.push(message);
+ links.set(key, link);
+ }
+ for (const { from, to, transport, messages: sent } of links.values()) {
+ const start = message => now + (message.at - first);
+ if (sent.length <= FLOWS_AT_MOST) {
+ for (const message of sent) {
+ flows.push({ from, to, transport, start: start(message), kind: message.counted,
+ r: dotRadius(message.bytes) });
}
- if (total === 0) continue;
- const shown = dotsFor(kinds, total);
- shown.forEach((dot, i) => {
- flows.push({ from, to, start: now + i * FLOW_SPREAD_MS / shown.length,
- kind: dot.kind, r: dotRadius(dot.bytes) });
- });
+ continue;
+ }
+ const kinds = {};
+ for (const message of sent) {
+ const kind = kinds[message.counted] || (kinds[message.counted] = { messages: 0, bytes: 0 });
+ kind.messages += 1;
+ kind.bytes += message.bytes;
}
+ const span = Math.max(...sent.map(message => message.at)) - Math.min(...sent.map(message => message.at));
+ const from0 = start(sent.reduce((a, b) => (a.at <= b.at ? a : b)));
+ const shown = dotsFor(kinds, sent.length);
+ shown.forEach((dot, i) => {
+ flows.push({ from, to, transport, start: from0 + i * span / shown.length,
+ kind: dot.kind, r: dotRadius(dot.bytes) });
+ });
}
if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
}
-// Whether dots are drawn now, for an update whose traffic came `gap` ms after the last:
-// in its time (not the first after a gap, the computer asleep or the stream stalled,
-// whose counts grew over all of it), shown, and seen.
-function dotsNow(gap) {
- return gap >= FLOW_SPREAD_MS / 4 && gap <= FLOW_SPREAD_MS * 2.5 && flowsShown &&
- !calm.matches && !document.hidden &&
- !document.getElementById("map").classList.contains("collapsed");
+// This node's messages, as its stream brings them (`messages` events): listed in the
+// console, and run along its links, those it sent, and those it received from a node
+// whose messages the page doesn't read (a subscribed node's run from it, as they left).
+function ownMessagesArrived(list) {
+ messagesArrived(list);
+ if (!state) return;
+ const self = state.node.id;
+ runMessages(list.filter(message => message.way === "sent" ||
+ (message.way === "received" && !logging(message.peer)))
+ .map(message => ({ ...message,
+ from: message.way === "sent" ? self : message.peer,
+ to: message.way === "sent" ? message.peer : self })));
}
-// A subscribed node's messages (`follow`), those of a tick: listed in the console and
+// A subscribed node's messages (`follow`), as they come: listed in the console and
// run along its links, as this node's own. Its messages with this node are this node's
// to show, and what it received from another subscribed node that one's (as what it
// sent): each message once. Each named by both ends (`from`, `to`) and the node that
@@ -3130,10 +3144,8 @@ function dotsNow(gap) {
function remoteMessagesArrived(id, entry, list) {
if (!state) return;
const self = state.node.id;
- const wall = Date.now(), gap = wall - (entry.messagesAt || 0);
- entry.messagesAt = wall;
- // A tick's messages come whole, those since the last, up to the latest 500: one
- // missing between (a busy tick) leaves what was delivered unknown for now.
+ // Its messages come whole, those since the last event, up to the latest 500: one
+ // missing between (a busy moment) leaves what was delivered unknown for now.
if (list.length > 0) {
if (entry.lastSeq !== undefined && list[0].seq !== entry.lastSeq + 1) forgetDeliveries(id);
entry.lastSeq = list[list.length - 1].seq;
@@ -3157,29 +3169,7 @@ function remoteMessagesArrived(id, entry, list) {
}
}
checkDeliveries();
- if (dotsNow(gap) && running.length > 0) {
- const now = performance.now();
- const links = new Map();
- for (const message of running) {
- const transport = schemeOf(message.address);
- const key = message.from + " " + message.to + " " + (transport || "");
- const link = links.get(key) ||
- { from: message.from, to: message.to, transport, kinds: {}, total: 0 };
- const kind = link.kinds[message.counted] || (link.kinds[message.counted] = { messages: 0, bytes: 0 });
- kind.messages += 1;
- kind.bytes += message.bytes;
- link.total += 1;
- links.set(key, link);
- }
- for (const { from, to, transport, kinds, total } of links.values()) {
- const shown = dotsFor(kinds, total);
- shown.forEach((dot, i) => {
- flows.push({ from, to, transport, start: now + i * FLOW_SPREAD_MS / shown.length,
- kind: dot.kind, r: dotRadius(dot.bytes) });
- });
- }
- if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
- }
+ runMessages(running);
if (kept.length > 0) messagesArrived(kept);
}
// Whether the page reads `id`'s messages one by one: a subscribed node's, while they
@@ -3646,7 +3636,7 @@ function connect() {
remakesArrived();
});
events.addEventListener("document", event => { pulse(); documentChanged(JSON.parse(event.data)); });
- events.addEventListener("messages", event => messagesArrived(JSON.parse(event.data)));
+ events.addEventListener("messages", event => ownMessagesArrived(JSON.parse(event.data)));
// The node stops: it says so, and ends the stream. Back once it is.
events.addEventListener("offline", () => {
events.close(); live = false;
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index d6485d7..1d63f38 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -37,6 +37,11 @@ use crate::swarm::{members, topology};
use crate::transport::Hex;
use links::Link;
+/// How long messages logged are gathered before they are sent on, after the first: a
+/// busy node sends about ten events a second at most, a quiet one each message soon
+/// after it went.
+const MESSAGES_GATHERED: Duration = Duration::from_millis(100);
+
/// How often the snapshot is built and compared, while anyone looks.
const TICK: Duration = Duration::from_secs(2);
/// How often the member list's graph is recomputed (it costs about `members²`).
@@ -91,8 +96,8 @@ struct Hub {
/// `{"paths": [...]}`, or `{"all": true}` if too many or not known. Shared by every page
/// open; each is told of the paths its user may read at least partly (`changed_for`).
documents: broadcast::Sender<Arc<String>>,
- /// The messages exchanged since the last tick, one by one (`traffic::MessageLog`),
- /// to the clients connected: a JSON list.
+ /// The messages exchanged, one by one (`traffic::MessageLog`), to the clients
+ /// connected as they come (`forward_messages`): a JSON list.
messages: broadcast::Sender<Arc<String>>,
control: Mutex<Control>,
/// Someone asked for it now (`refresh`): a tick at once, the graph recomputed.
@@ -168,10 +173,11 @@ impl Hub {
let mut graph = Graph::default();
let mut graph_due = Instant::now();
let mut changes = self.state.change_events.subscribe();
- // The messages one by one, from now on, while it runs.
+ // The messages one by one, from now on, while it runs: sent on as they come.
let log = &self.state.traffic.log;
log.watch(true);
- let mut logged = 0u64;
+ let forwarding = tokio::spawn(Arc::clone(&self).forward_messages());
+ let _forwarding = AbortOnDrop(forwarding);
loop {
if Instant::now() >= graph_due {
graph.refresh(&self.state);
@@ -204,14 +210,6 @@ impl Hub {
if let Some(changed) = document_changes(&mut changes) {
let _ = self.documents.send(Arc::new(changed.to_string()));
}
- let mut messages = log.after(logged, MESSAGES_AT_MOST);
- by_node(&self.state, &mut messages).await;
- if let Some(last) = messages.last() {
- logged = last.seq;
- if let Ok(json) = serde_json::to_string(&messages) {
- let _ = self.messages.send(Arc::new(json));
- }
- }
let unwanted = || {
self.patches.receiver_count() == 0
&& Instant::now() >= self.control.lock_or_recover().wanted_until
@@ -236,6 +234,38 @@ impl Hub {
}
}
+impl Hub {
+ /// The messages logged, to every client connected (`messages`), as they come:
+ /// gathered for `MESSAGES_GATHERED` after the first, the latest `MESSAGES_AT_MOST`.
+ /// For as long as the producer runs.
+ async fn forward_messages(self: Arc<Self>) {
+ let log = &self.state.traffic.log;
+ let mut logged = log.latest();
+ loop {
+ log.logged_after(logged).await;
+ tokio::time::sleep(MESSAGES_GATHERED).await;
+ let mut messages = log.after(logged, MESSAGES_AT_MOST);
+ let Some(last) = messages.last() else {
+ continue;
+ };
+ logged = last.seq;
+ by_node(&self.state, &mut messages).await;
+ if let Ok(json) = serde_json::to_string(&messages) {
+ let _ = self.messages.send(Arc::new(json));
+ }
+ }
+ }
+}
+
+/// A task stopped when this is dropped.
+struct AbortOnDrop(tokio::task::JoinHandle<()>);
+
+impl Drop for AbortOnDrop {
+ fn drop(&mut self) {
+ self.0.abort();
+ }
+}
+
/// Paths changed in a tick listed at most; more: `all`.
/// The messages the paths layer logged under an address (its own PINGs and PONGs) under
/// the node there instead, where one is known: the address it proved itself at, or is
@@ -536,9 +566,9 @@ impl Drop for Watching {
}
/// `GET /admin/dashboard/messages/own/events`: server-sent events, `messages` with the
-/// messages this node sent and received since the last (every `TICK`, those of the
-/// tick, none too, as the page's own `messages` events have them, without each message
-/// written out), for a page of another member subscribed to this node: its console lists them,
+/// messages this node sent and received since the last, as they come (gathered for
+/// `MESSAGES_GATHERED` after the first), as the page's own `messages` events have them,
+/// without each message written out, for a page of another member subscribed to this node: its console lists them,
/// and its map runs them along this node's links. `offline` when the node stops, which
/// ends it. The message log is kept while it is open.
async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
@@ -548,21 +578,25 @@ async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoRe
let mut logged = log.latest();
let stream = async_stream::stream! {
let _watching = watching;
- let mut every = tokio::time::interval(TICK);
loop {
+ // As they come, gathered a moment (`MESSAGES_GATHERED`).
+ let log = &hub.state.traffic.log;
tokio::select! {
- _ = every.tick() => {}
+ () = async {
+ log.logged_after(logged).await;
+ tokio::time::sleep(MESSAGES_GATHERED).await;
+ } => {}
() = hub.closed() => {
yield Ok::<_, Infallible>(offline_event());
break;
}
}
- // Every tick, none too: the page times its dots by them.
- let mut messages = hub.state.traffic.log.after(logged, MESSAGES_AT_MOST);
- if let Some(last) = messages.last() {
- logged = last.seq;
- by_node(&hub.state, &mut messages).await;
- }
+ let mut messages = log.after(logged, MESSAGES_AT_MOST);
+ let Some(last) = messages.last() else {
+ continue;
+ };
+ logged = last.seq;
+ by_node(&hub.state, &mut messages).await;
let messages: Vec<JsonValue> = messages
.into_iter()
.filter_map(|message| {