From 8e5dda246c4684908613178034014e79537ebfc2 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 9 Nov 2026 00:02:33 +0000 Subject: dashboard: messages as they come The node sent its messages to the page every 2 s (`messages` events, this node's and a subscribed node's), and this node's ran on the map from its counts, which come with the state: the map showed a message 2 to 4 s after it went, the messages of each 2 s spread evenly over the next. The message log now tells those waiting of each message logged (MessageLog::logged_after), and both streams send what was logged a tenth of a second after the first of it (MESSAGES_GATHERED): a busy node about ten events a second, a quiet one each message soon after it went. The page runs messages as their streams bring them, this node's from its own log too (with the addresses its PINGs went to), as far apart as they were sent; more than six on a link and way at once are shared out by kind over the moment, and after a gap only the last 2 s run. The counts still give the rates and what was delivered. --- docs/dashboard.md | 46 ++++++------ src/api/dashboard/dashboard.html | 154 ++++++++++++++++++--------------------- src/api/dashboard/mod.rs | 80 ++++++++++++++------ src/transport/traffic.rs | 14 ++++ 4 files changed, 167 insertions(+), 127 deletions(-) diff --git a/docs/dashboard.md b/docs/dashboard.md index 54aefa7..0b1b10d 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -137,32 +137,33 @@ drawn hollow) is drawn as the other end reports it, which its tooltip says. Then Messages this node exchanges run along its links as dots, from sender to receiver, all at one speed: over a longer link they arrive later. Each follows the line it went over: -a subscribed node's PING or PONG of the paths layer, which names the address it went -to, the line of that address's transport (curved, beside the one in use); any other -(this node's own come as counts, without addresses), the line in use. -They come from the per-peer message counts, which arrive every 2 s: what was sent and -received since is spread over the next 2 s, at most 6 dots per link and direction, -shared out by message kind, each kind in its colour (named in the legend; the console -colours them alike). Counts that come after a gap (the computer slept, the stream -stalled), and those that were queued meanwhile and come at once with them, show no -dots: they would all run at once. A dot is as large as its kind's messages were on average -(by the size's logarithm: a ping small, a large sync message large); the `size` slider +a PING or PONG of the paths layer, which names the address it went to, the line of +that address's transport (curved, beside the one in use); any other, the line in use. +They come from the node's message log as it goes: the node sends what it logged a tenth +of a second after the first of them, so a message runs soon after it went, and the +messages of a moment start as far apart as they were sent. Of more than 6 in a moment +on a link and way, 6 run, shared out by message kind, spread over the moment; each kind +in its colour (named in the legend; the console colours them alike). After a gap (the +computer slept, the stream stalled), only the last 2 s of what then comes runs. A dot +is as large as its message (by the size's logarithm: a ping small, a large sync message +large; shared out, as its kind's messages on average); the `size` slider right after their legend makes all dots larger or smaller, and the `speed` slider after it makes them run faster or slower (0.1 to 0.9 map units per millisecond, 0.3 at first: a link of 200 in about 0.7 s); both are kept in the browser. A picture of the traffic, not a record of it. A subscribed node's messages with its peers run -along its links too, from its messages of each 2 s (read from it with the page's token, -as its links are). Each message runs once, as it left, from its sender where the page -reads the sender (this node's counts, or a subscribed node's messages): so it runs the +along its links too, from its message log as it goes (read from it with the page's +token, as its links are). Each message runs once, as it left, from its sender where the +page reads the sender (this node's, or a subscribed node's): so it runs the way it went, also where it never arrived. Only a message from a node the page doesn't read runs as it arrived, from its receiver's. Links between other members show none. Where the page reads both ends of a link, what one sent the other should count received: a message sent and not received within 5 s makes the link breathe red once, its tooltip saying how many didn't arrive. Every kind counts but the paths and -identity layers' (PINGs, WHO), which this node counts under addresses, not nodes; a -tick of a subscribed node's messages that came short (more than 500 in 2 s) starts the -count afresh, so it never takes what it didn't see as lost. +identity layers' (PINGs, WHO), which this node counts under addresses, not nodes. This +node's side is its counts, which come with the state every 2 s, hence the 5 s; a +subscribed node's messages that came short (more than 500 at once) start the count +afresh, so it never takes what it didn't see as lost. `messages: dots` after the sliders switches to `messages: flash`: each message crosses its link as fast as it did (half the link's round trip, as the map has it; 50 ms where none is measured), a streak widening to its head, which then draws back into where it @@ -350,8 +351,8 @@ clock, to the millisecond), which way (`→` sent, `←` received, `↛` refused transport wouldn't send it), with whom, the type (`HELLO`, `SYNC`, `STATUS_REPLY`, `PING`, …, in its kind's colour, as on the map) and the size. The latest 500, the newest at the bottom; scrolled along while at the bottom, which `following…` beside the -title says. `clear` empties it; what comes from then on fills it again. They arrive with every update -(every 2 s). The PINGs the paths layer sends on its own, and the PONGs it answers +title says. `clear` empties it; what comes from then on fills it again. They arrive as +they come (a tenth of a second after the first of a moment's). The PINGs the paths layer sends on its own, and the PONGs it answers PINGs with, are listed too: under the node at the address they went to, where it is known, else under the address. Not listed: WHO and ME, which stay in the layers below the node (they are counted, [`measurements.md`](measurements.md), Traffic). @@ -464,10 +465,11 @@ takes each against their median: a member more than `--clock-skew-warn` off it, beyond the chained bound, is ringed. With fewer than three members compared, none is. -**Messages on the map** come from the per-peer message counts in the state (`/peers`, -`traffic`), not from messages of their own. **The console's** come from a log the node -keeps only while the producer runs (`traffic::MessageLog`, the latest 1000), sent as a -`messages` event every 2 s with those since the last. +**Messages**, on the map and in the console, come from a log the node keeps only while +someone looks (`traffic::MessageLog`, the latest 1000), sent as a `messages` event as +they come: a tenth of a second after the first of those since the last, the latest 500 +of them. The per-peer message counts in the state (`/peers`, `traffic`) give the rates +and what was delivered. **The document** is read through the store API, a level at a time as it is opened (`GET /store?shallow=true`, then `/store/?shallow=true` for each map opened, and 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>, - /// 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>, control: Mutex, /// 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) { + 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>) -> impl IntoResponse { @@ -548,21 +578,25 @@ async fn own_messages_events(Extension(hub): Extension>) -> 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 = messages .into_iter() .filter_map(|message| { diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index b9506a9..948e5d5 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -158,6 +158,8 @@ const LOGGED_TEXT: usize = 8 * 1024; pub(crate) struct MessageLog { /// How many look (`watch`). watching: AtomicUsize, + /// Told of each message logged, for those sending them on as they come (`logged`). + logged: tokio::sync::Notify, inner: Mutex, } @@ -293,6 +295,18 @@ impl MessageLog { if inner.entries.len() > LOGGED { inner.entries.pop_front(); } + drop(inner); + self.logged.notify_waiters(); + } + + /// Until a message after `seq` is logged: at once if one is. Blocking nothing. + pub(crate) async fn logged_after(&self, seq: u64) { + // Made before looking, so a message logged in between still wakes it. + let logged = self.logged.notified(); + if self.latest() > seq { + return; + } + logged.await; } /// The messages after `seq`, the latest `most` of them. -- cgit v1.3.1