From 2442525bc93fbb55b156549389f3c4b25ffdb1dd Mon Sep 17 00:00:00 2001 From: rdyplayerB Date: Tue, 22 Sep 2026 15:22:28 -0700 Subject: [PATCH 1/2] feat(coordinator): feed the CUDA miner from a quip-screen sidecar With QUIP_SCREEN_BIN set, the feeder writes the snapshot topology to a spec file, runs `quip-screen serve`, sends it each round's prev hash and identity, and queues the salts it keeps, deepest first per generation. cuda* miners are topped up from that queue; the job is still derived with derive_pow_job, so validation, stash and submission are unchanged. An empty queue stages nothing unless QUIP_SCREEN_FALLBACK=1; a dead sidecar falls back to counter salts. Unset: no behaviour change. --- COORDINATOR.md | 15 + crates/quip-coordinator/src/lib.rs | 1 + crates/quip-coordinator/src/runtime.rs | 22 +- crates/quip-coordinator/src/screen.rs | 368 +++++++++++++++++++++++++ 4 files changed, 404 insertions(+), 2 deletions(-) create mode 100644 crates/quip-coordinator/src/screen.rs diff --git a/COORDINATOR.md b/COORDINATOR.md index 6fa3bb9d..f0faa93e 100644 --- a/COORDINATOR.md +++ b/COORDINATOR.md @@ -246,6 +246,21 @@ When `last_proof_block_hash` changes, the feeder drives the round state machine. Startup drives the same machine. Mining is the last state. A new qblock head in any state returns the machine to the first state. +### Probe screen (optional) + +With `QUIP_SCREEN_BIN` pointing at `quip-screen` (quip-miner-cuda), the feeder +starts it as a sidecar on the first poll. It writes the snapshot's topology to +`$TMPDIR/quip-screen-.json`, passes `--spec` and `QUIP_SCREEN_ARGS`, and +sends `R ` on every round change. The +sidecar probes fresh salts with the same `derive_nonce` and prints the deepest +as `K `. When topping up a `cuda*` miner, the +feeder pops the deepest queued salt of the current generation and derives the +job from it with `derive_pow_job`, so validation, stash and submission do not +change. With the queue empty it stages nothing for that miner, which leaves the +GPU to the screen; `QUIP_SCREEN_FALLBACK=1` tops it up with counter salts +instead. If the sidecar exits, the feeder uses counter salts. A topology change +restarts the sidecar with a new spec. Module: `screen.rs`. + ### Round state machine The states, in order: diff --git a/crates/quip-coordinator/src/lib.rs b/crates/quip-coordinator/src/lib.rs index 851d402d..0d49cb1a 100644 --- a/crates/quip-coordinator/src/lib.rs +++ b/crates/quip-coordinator/src/lib.rs @@ -20,6 +20,7 @@ pub mod readiness; pub mod round; pub mod router; pub mod runtime; +pub mod screen; pub mod session; pub mod stash; pub mod supervisor; diff --git a/crates/quip-coordinator/src/runtime.rs b/crates/quip-coordinator/src/runtime.rs index 7f621164..74243f58 100644 --- a/crates/quip-coordinator/src/runtime.rs +++ b/crates/quip-coordinator/src/runtime.rs @@ -987,6 +987,11 @@ pub async fn feeder_loop( // and emitted off-lock below. The completion pair is (window, // total): window is completions since the last heartbeat. let mut stats: Vec<(String, u64, u64, usize, usize)> = Vec::new(); + // Probe screen: tell the sidecar which round to screen. + let screen_on = crate::screen::enabled(); + if screen_on { + crate::screen::announce(generation, &snap, params.miner_identity); + } for id in st.router.miner_ids() { let consumed_raw = st.router.take_consumed(&id); let consumed = f64::from(consumed_raw); @@ -1005,8 +1010,21 @@ pub async fn feeder_loop( params.buffer_depth, )); while st.router.staged_len(&id) < depth { - salt_ctr = salt_ctr.saturating_add(1); - let salt = salt_from_counter(salt_ctr); + // Probe screen: the CUDA miner gets the deepest screened + // salt; counter salts only as the configured fallback. + let salt = if screen_on && id.starts_with("cuda") { + match crate::screen::pop(generation) { + Some(s) => s, + None if crate::screen::fallback() => { + salt_ctr = salt_ctr.saturating_add(1); + salt_from_counter(salt_ctr) + } + None => break, + } + } else { + salt_ctr = salt_ctr.saturating_add(1); + salt_from_counter(salt_ctr) + }; let job = match derive_pow_job( &snap, params.miner_identity, diff --git a/crates/quip-coordinator/src/screen.rs b/crates/quip-coordinator/src/screen.rs new file mode 100644 index 00000000..a2ac0e67 --- /dev/null +++ b/crates/quip-coordinator/src/screen.rs @@ -0,0 +1,368 @@ +//! Probe-screen feed for the CUDA miner. +//! +//! A `PoW` instance is drawn from its nonce, and the instance sets almost all of +//! the energy a solve can reach. `quip-screen` (quip-miner-cuda) anneals many +//! fresh nonces briefly on the GPU and reports the deepest; the feeder then +//! gives the CUDA miner those salts instead of counter salts, so its full solve +//! is spent on instances that are likely to clear the target. +//! +//! The sidecar derives nonces exactly as the feeder does +//! (`derive_nonce(last_proof_block_hash, miner_identity, salt)`), and the feeder +//! derives every job from a screened salt with the ordinary +//! `derive_pow_job`, so validation, stash and submission are +//! unchanged. +//! +//! Environment: +//! - `QUIP_SCREEN_BIN`: path to `quip-screen`. Unset or empty: off. +//! - `QUIP_SCREEN_ARGS`: extra `quip-screen` arguments, whitespace separated, +//! passed after `--spec ` (e.g. `--sweeps 512 serve --keep-per-s 40`). +//! `serve` is appended when absent. +//! - `QUIP_SCREEN_FALLBACK=1`: top the CUDA miner up with counter salts while no +//! screened salt is queued (default: leave the GPU to the screen). +//! +//! Protocol (line based): the feeder writes `R +//! ` when the round changes; the sidecar writes `K +//! ` per kept nonce and `S +//! ...` every 10 s. Closing stdin stops the sidecar. If the sidecar +//! exits, the feeder falls back to counter salts. + +use crate::chain::snapshot::MiningSnapshot; +use std::cmp::Reverse; +use std::collections::BinaryHeap; +use std::io::{BufRead, BufReader, Write}; +use std::path::PathBuf; +use std::process::{ChildStdin, Command, Stdio}; +use std::sync::{Mutex, MutexGuard, OnceLock}; + +/// Screened salts held per round; the deepest are served first. +const QUEUE_CAP: usize = 20_000; + +/// Screened salts of the current round, deepest first. +#[derive(Debug, Default)] +pub struct SaltQueue { + generation: u64, + heap: BinaryHeap<(Reverse, [u8; 32])>, +} + +impl SaltQueue { + /// Start a new round: drop every salt of the previous one. + pub fn reset(&mut self, generation: u64) { + self.generation = generation; + self.heap.clear(); + } + + /// Queue a salt reported for `generation`. Salts of another round, or past + /// the cap, are dropped. Returns whether it was queued. + pub fn push(&mut self, generation: u64, salt: [u8; 32], energy: i64) -> bool { + if generation != self.generation || self.heap.len() >= QUEUE_CAP { + return false; + } + self.heap.push((Reverse(energy), salt)); + true + } + + /// The deepest queued salt of `generation`. + pub fn pop(&mut self, generation: u64) -> Option<([u8; 32], i64)> { + if generation != self.generation { + return None; + } + self.heap.pop().map(|(Reverse(e), salt)| (salt, e)) + } + + /// Salts waiting. + #[must_use] + pub fn len(&self) -> usize { + self.heap.len() + } + + /// Whether no salt is waiting. + #[must_use] + pub fn is_empty(&self) -> bool { + self.heap.is_empty() + } +} + +#[derive(Default)] +struct Sidecar { + stdin: Option, + alive: bool, + topology: Vec, + announced: Option<(u64, [u8; 32])>, + queue: SaltQueue, + received: u64, + fed: u64, + best_fed: Option, + /// Set when the sidecar could not start or exited on its own: counter + /// salts from then on, no respawn loop. + failed: bool, + /// Bumped per spawn, so a reader thread only reports its own sidecar's exit. + epoch: u64, +} + +fn sidecar() -> MutexGuard<'static, Sidecar> { + static S: OnceLock> = OnceLock::new(); + S.get_or_init(|| Mutex::new(Sidecar::default())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn bin() -> Option<&'static str> { + static B: OnceLock> = OnceLock::new(); + B.get_or_init(|| { + std::env::var("QUIP_SCREEN_BIN") + .ok() + .filter(|v| !v.trim().is_empty()) + }) + .as_deref() +} + +/// Whether the screen is configured. +#[must_use] +pub fn enabled() -> bool { + bin().is_some() +} + +/// Whether the CUDA miner should get counter salts while the queue is empty: +/// `QUIP_SCREEN_FALLBACK=1`, or no sidecar running. +#[must_use] +pub fn fallback() -> bool { + static F: OnceLock = OnceLock::new(); + *F.get_or_init(|| std::env::var("QUIP_SCREEN_FALLBACK").is_ok_and(|v| v == "1")) + || !sidecar().alive +} + +fn hex(bytes: &[u8]) -> String { + use std::fmt::Write as _; + bytes + .iter() + .fold(String::with_capacity(bytes.len() * 2), |mut s, b| { + // Writing to a String cannot fail. + let _ = write!(s, "{b:02x}"); + s + }) +} + +fn unhex32(s: &str) -> Option<[u8; 32]> { + if s.len() != 64 { + return None; + } + let mut out = [0u8; 32]; + for (i, byte) in out.iter_mut().enumerate() { + *byte = u8::from_str_radix(s.get(2 * i..2 * i + 2)?, 16).ok()?; + } + Some(out) +} + +/// The snapshot's topology as the drive/seed-chain spec JSON `quip-screen` +/// reads: nodes, edges in chain order, and the allowed values. +#[must_use] +pub fn spec_json(snap: &MiningSnapshot) -> String { + serde_json::json!({ + "nodes": snap.nodes, + "edges": snap.edges.iter().map(|&(u, v)| [u, v]).collect::>(), + "allowed_h_milli": snap.allowed_h_milli, + "allowed_j_milli": snap.allowed_j_milli, + }) + .to_string() +} + +/// Parse one sidecar line into the queue. Returns the stats line fields, if +/// it was one. +fn ingest(line: &str) -> Option { + let parts: Vec<&str> = line.split_whitespace().collect(); + match parts.as_slice() { + ["K", generation, salt, energy] => { + if let (Ok(g), Some(salt), Ok(e)) = ( + generation.parse::(), + unhex32(salt), + energy.parse::(), + ) { + let mut s = sidecar(); + s.received += 1; + let _queued = s.queue.push(g, salt, e); + } + None + } + ["S", rest @ ..] => Some(rest.join(" ")), + _ => None, + } +} + +fn spawn(s: &mut Sidecar, snap: &MiningSnapshot) { + let Some(path) = bin() else { return }; + let spec: PathBuf = std::env::temp_dir().join(format!( + "quip-screen-{}.json", + hex(snap.topology_hash.get(..8).unwrap_or(&snap.topology_hash)) + )); + if let Err(e) = std::fs::write(&spec, spec_json(snap)) { + tracing::error!(error = %e, path = %spec.display(), "screen: cannot write the topology spec; counter salts only"); + s.failed = true; + return; + } + let extra: Vec = std::env::var("QUIP_SCREEN_ARGS") + .unwrap_or_default() + .split_whitespace() + .map(str::to_owned) + .collect(); + let mut args = vec!["--spec".to_owned(), spec.display().to_string()]; + let has_serve = extra.iter().any(|a| a == "serve"); + args.extend(extra); + if !has_serve { + args.push("serve".to_owned()); + } + let child = Command::new(path) + .args(&args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn(); + let mut child = match child { + Ok(c) => c, + Err(e) => { + tracing::error!(error = %e, bin = path, "screen: cannot start the sidecar; counter salts only"); + s.failed = true; + return; + } + }; + let (Some(stdin), Some(stdout)) = (child.stdin.take(), child.stdout.take()) else { + tracing::error!("screen: sidecar pipes missing; counter salts only"); + s.failed = true; + return; + }; + s.epoch += 1; + let epoch = s.epoch; + s.stdin = Some(stdin); + s.alive = true; + s.topology.clone_from(&snap.topology_hash); + s.announced = None; + tracing::info!(bin = path, ?args, "screen: sidecar started"); + let reader = std::thread::Builder::new() + .name("screen-reader".into()) + .spawn(move || { + for line in BufReader::new(stdout).lines() { + let Ok(line) = line else { break }; + if let Some(stats) = ingest(&line) { + let mut s = sidecar(); + tracing::info!( + sidecar = %stats, + received = s.received, + fed = s.fed, + queued = s.queue.len(), + best_fed = ?s.best_fed, + "screen stats" + ); + s.best_fed = None; + } + } + let current = { + let mut s = sidecar(); + let current = s.epoch == epoch; + if current { + s.alive = false; + s.stdin = None; + s.failed = true; + } + current + }; + if current { + tracing::error!("screen: sidecar exited; counter salts only"); + } + if let Err(e) = child.wait() { + tracing::warn!(error = %e, "screen: cannot reap the sidecar"); + } + }); + if let Err(e) = reader { + tracing::error!(error = %e, "screen: cannot start the reader thread"); + s.alive = false; + s.stdin = None; + s.failed = true; + } +} + +/// Tell the sidecar which round to screen, starting it on first use. A +/// topology change restarts it with a fresh spec. Cheap when nothing changed. +pub fn announce(generation: u64, snap: &MiningSnapshot, miner_identity: [u8; 32]) { + let mut s = sidecar(); + if s.alive && s.topology != snap.topology_hash { + tracing::info!("screen: topology changed; restarting the sidecar"); + // Closing stdin ends the sidecar; its reader thread then marks it dead. + s.stdin = None; + s.alive = false; + s.failed = false; + } + if !s.alive && !s.failed && s.stdin.is_none() { + spawn(&mut s, snap); + } + if s.announced == Some((generation, snap.last_proof_block_hash)) { + return; + } + s.queue.reset(generation); + s.announced = Some((generation, snap.last_proof_block_hash)); + let line = format!( + "R {generation} {} {}\n", + hex(&snap.last_proof_block_hash), + hex(&miner_identity) + ); + let broken = s.stdin.as_mut().is_some_and(|w| { + w.write_all(line.as_bytes()) + .and_then(|()| w.flush()) + .is_err() + }); + if broken { + s.stdin = None; + s.alive = false; + } +} + +/// The deepest screened salt queued for `generation`, if any. +#[must_use] +pub fn pop(generation: u64) -> Option<[u8; 32]> { + let mut s = sidecar(); + let (salt, energy) = s.queue.pop(generation)?; + s.fed += 1; + s.best_fed = Some(s.best_fed.map_or(energy, |b| b.min(energy))); + Some(salt) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn queue_serves_deepest_first_within_its_round() { + let mut q = SaltQueue::default(); + q.reset(7); + assert!(q.push(7, [1; 32], -14_400)); + assert!(q.push(7, [2; 32], -14_500)); + assert!(!q.push(6, [3; 32], -14_900), "a stale round is dropped"); + assert_eq!(q.pop(8), None, "a newer round sees nothing"); + assert_eq!(q.pop(7), Some(([2; 32], -14_500))); + assert_eq!(q.pop(7), Some(([1; 32], -14_400))); + assert!(q.is_empty()); + } + + #[test] + fn reset_drops_the_previous_round() { + let mut q = SaltQueue::default(); + q.reset(1); + assert!(q.push(1, [1; 32], -1)); + q.reset(2); + assert!(q.is_empty()); + } + + #[test] + fn hex_round_trips() { + let b: [u8; 32] = core::array::from_fn(|i| u8::try_from(i * 7 % 256).unwrap_or(0)); + assert_eq!(unhex32(&hex(&b)), Some(b)); + assert_eq!(unhex32("zz"), None); + } + + #[test] + fn stats_lines_are_passed_through_and_salts_parsed() { + assert_eq!( + ingest("S 24000 -14397 40.1 r4x512").as_deref(), + Some("24000 -14397 40.1 r4x512") + ); + assert_eq!(ingest("K not-a-number 00 1"), None); + } +} From 6ab294057545561f02b41674272ef2747a56a478 Mon Sep 17 00:00:00 2001 From: rdyplayerB Date: Tue, 22 Sep 2026 17:45:45 -0700 Subject: [PATCH 2/2] fix(coordinator): keep the deepest half of a full screen queue In a long round the screened-salt queue reached its 20,000 cap and then dropped every new salt, including ones deeper than those waiting. At the cap it now keeps its deepest half and admits the new salt. --- crates/quip-coordinator/src/screen.rs | 39 ++++++++++++++++++++++++--- 1 file changed, 36 insertions(+), 3 deletions(-) diff --git a/crates/quip-coordinator/src/screen.rs b/crates/quip-coordinator/src/screen.rs index a2ac0e67..e1df3145 100644 --- a/crates/quip-coordinator/src/screen.rs +++ b/crates/quip-coordinator/src/screen.rs @@ -51,12 +51,20 @@ impl SaltQueue { self.heap.clear(); } - /// Queue a salt reported for `generation`. Salts of another round, or past - /// the cap, are dropped. Returns whether it was queued. + /// Queue a salt reported for `generation`; salts of another round are + /// dropped. At the cap the queue keeps its deepest half, so a long round + /// still admits later, deeper salts. Returns whether it was queued. pub fn push(&mut self, generation: u64, salt: [u8; 32], energy: i64) -> bool { - if generation != self.generation || self.heap.len() >= QUEUE_CAP { + if generation != self.generation { return false; } + if self.heap.len() >= QUEUE_CAP { + // Ascending by `Reverse(energy)` is shallowest first. + let mut kept = std::mem::take(&mut self.heap).into_sorted_vec(); + kept.reverse(); + kept.truncate(QUEUE_CAP / 2); + self.heap = kept.into_iter().collect(); + } self.heap.push((Reverse(energy), salt)); true } @@ -341,6 +349,31 @@ mod tests { assert!(q.is_empty()); } + #[test] + fn a_full_queue_keeps_its_deepest_half_and_admits_new_salts() { + let mut q = SaltQueue::default(); + q.reset(1); + for i in 0..QUEUE_CAP { + let e = -i64::try_from(i).unwrap_or(0); + assert!(q.push(1, [0; 32], e)); + } + assert!( + q.push(1, [9; 32], -1_000_000), + "a deeper salt is admitted when full" + ); + assert_eq!(q.len(), QUEUE_CAP / 2 + 1); + assert_eq!(q.pop(1), Some(([9; 32], -1_000_000))); + let shallowest_kept = -i64::try_from(QUEUE_CAP / 2).unwrap_or(0) + 1; + let mut last = i64::MIN; + while let Some((_, e)) = q.pop(1) { + assert!( + e >= last && e <= shallowest_kept, + "deepest half, served deepest first" + ); + last = e; + } + } + #[test] fn reset_drops_the_previous_round() { let mut q = SaltQueue::default();