Multi-host prefix cache: startup row thinning of a recovered disk store overflows the 64-frame record backlog and stops the server

Aperta
  • Qualcosa non va?
#189 1 commento 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
3/5
Tempo stimato
1-2 giorni
Idoneità per principianti
42/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
rust
Ambito
backend, performance

Direzione di ricerca

Start with thin_rows and maintenance scheduling in selene/src/disk_store.rs, then inspect the backlog handling in selene/src/shard.rs. Reproduce with the supplied recovered-chain test and compare the short and long cases. Done when startup thinning completes without overflowing the peer record backlog and the regression test passes.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

area:caching area:selene bug model:deepseek-v4.1 priority:P1 triage:investigate

Symptom. A TP3 server (one rank per host) whose ranks reuse an existing --prefix-cache-disk directory joins every rank ("4890 of 4890 records on every rank") and starts the HTTP server, then stops immediately:

  • rank 0: native replay did not drain: rank 2's prefix record connection failed while jobs were outstanding: prefix record connection backlog is full; the peer is not keeping up
  • peer: host prefix record service did not drain: prefix record connection closed

The same package with an empty directory starts and serves normally.

Root cause.

  • DiskStore::new seeds thinning with the parent of every scanned record (selene/src/disk_store.rs:1561).
  • The first poll calls thin_rows (disk_store.rs:2075, from disk_store.rs:2617 / composite.rs:1324), which compacts every redundant row in one pass with no bound. Each compaction is a truncate (disk_store.rs:2299).
  • With Sharded, each truncate is queued to every peer through Link::enqueue (selene/src/shard.rs:587-591, 897-903), which try_sends into a sync_channel(BACKLOG = 64) (shard.rs:29, 202, 239-266). A full queue marks the link broken and abandon poisons the transport (shard.rs:756).
  • The 64-frame bound is on the channel to the link's writer thread, so a single poll that queues more than about 64 frames for a peer can overflow before the writer thread drains it, even when the peer reads promptly over loopback.
  • A directory written before eager thinning (#168) can hold thousands of thinnable interior rows. Ours had 3485 per rank.

Conditions. A multi-host package with --prefix-cache-disk and a recovered directory holding more than about 64 interior records whose rows are redundant (each has one ready child). Not verified: any other path that queues more than 64 frames for a peer within one poll would fail the same way, for example make_room evicting many small leaves for one large store. Max-age expiry is capped at 64 per sweep (shared/index.rs:406), so it could only contribute together with other jobs. A peer that briefly stops reading makes all of these more likely.

Reproduction (CPU only). In the disk_store.rs tests, build a DiskStore over a scanned linear chain of N records, with the files present in both directories, and give it a Sharded transport with one Link to a crate::shard::serve(Peer { .. }) thread over a loopback TcpStream; then call store.poll() in a loop. With cargo test --release, N = 3500 fails 5 out of 5 runs after 65-80 truncates with the two errors above; N = 40-70 passes, and N = 100-400 fails intermittently.

Test (append to the mod tests of selene/src/disk_store.rs) 
    /// Repro: a store recovered over a linear chain thins every interior row in
    /// one `poll`, and over the multi-host transport each truncate becomes one
    /// queued frame per peer. With a local peer that reads promptly, a chain
    /// longer than `shard::BACKLOG` still poisons the transport.
    fn thin_recovered_chain_over_sharded(records: u64) -> std::result::Result<(), String> {
        use crate::shard::{Link, Peer, Sharded};
        let geometry = geometry();
        let full = history(records * PAGE_TOKENS);
        let zero_dir = std::env::temp_dir().join(format!("flare-repro-zero-{records}-{}", std::process::id()));
        let peer_dir = std::env::temp_dir().join(format!("flare-repro-peer-{records}-{}", std::process::id()));
        for d in [&zero_dir, &peer_dir] {
            let _ = fs::remove_dir_all(d);
            fs::create_dir_all(d).unwrap();
        }
        let mut scanned_records = Vec::new();
        for i in 1..=records {
            let parent = (i > 1).then(|| (i - 1) * PAGE_TOKENS);
            let header = header(&geometry, &full, i * PAGE_TOKENS, parent);
            let length = header.length_with_row();
            for d in [&zero_dir, &peer_dir] {
                let file = fs::File::create(d.join(record::file_name(&header.key))).unwrap();
                file.set_len(length).unwrap();
            }
            scanned_records.push(scanned(&zero_dir, header, i, true));
        }
        let staged = || {
            Staged::new(
                None,
                StoppedCopier::shared(&geometry).unwrap(),
                DiskThread::start("flare-repro").unwrap(),
            )
        };
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let client = std::net::TcpStream::connect(listener.local_addr().unwrap()).unwrap();
        let server = listener.accept().unwrap().0;
        let peer = Peer {
            staged: staged(),
            directory: peer_dir.clone(),
            fingerprint: FINGERPRINT,
            geometry: geometry.clone(),
            link: Link::new(server).unwrap(),
            _lock: crate::testing::directory_lock(&peer_dir).unwrap(),
        };
        let service = std::thread::spawn(move || crate::shard::serve(peer));
        let transport = Sharded::new(staged(), vec![Link::new(client).unwrap()]).unwrap();
        let mut store = DiskStore::new(
            crate::testing::directory_lock(&zero_dir).unwrap(),
            IDENTITY,
            FINGERPRINT,
            1 << 40,
            geometry.clone(),
            scanned_records,
            Box::new(transport),
        )
        .unwrap();
        let mut outcome = Ok(());
        let deadline = Instant::now() + Duration::from_secs(20);
        while Instant::now() < deadline {
            if let Err(error) = store.poll(&mut Vec::new()) {
                outcome = Err(error.to_string());
                break;
            }
            if store.stats().rows_thinned[1] == records - 2 && store.maintenance.is_empty() {
                break;
            }
            std::thread::sleep(Duration::from_millis(1));
        }
        let thinned = store.stats().rows_thinned[1];
        let pending = store.maintenance.len();
        let closed = store.close();
        drop(store);
        let served = service.join().unwrap();
        eprintln!(
            "records={records} thinned={thinned} pending_maintenance={pending} outcome={outcome:?} close={closed:?} peer={served:?}"
        );
        if outcome.is_ok() && pending != 0 {
            outcome = Err(format!("{pending} truncates did not complete"));
        }
        let _ = fs::remove_dir_all(&zero_dir);
        let _ = fs::remove_dir_all(&peer_dir);
        outcome
    }

    #[test]
    fn repro_short_recovered_chain_thins_over_sharded_transport() {
        // 40 records: 38 interior rows to thin, under the 64-frame backlog.
        thin_recovered_chain_over_sharded(std::env::var("REPRO_SHORT").ok().and_then(|v| v.parse().ok()).unwrap_or(40)).unwrap();
    }

    #[test]
    fn repro_long_recovered_chain_thinning_overflows_the_record_backlog() {
        // 3500 records (as in the failing directory): 3498 truncates queued for the peer in one poll.
        let error = thin_recovered_chain_over_sharded(3500).unwrap_err();
        assert!(error.contains("backlog is full"), "{error}");
    }

Suggested fix, one or more of:

  1. Bound thin_rows (and other maintenance loops) per poll: stop issuing truncates while maintenance.len() is at or above a burst limit below BACKLOG, and keep the remaining keys in thinning. Locally, a limit of 32 makes N = 3500 (and 100, 400) pass in about 0.5 s and keeps the 38 existing disk_store tests passing.
  2. In Sharded, keep a per-peer overflow queue and flush it into the link on each poll instead of failing on TrySendError::Full, so the bounded channel limits memory rather than correctness; real stalls are still caught by JOB_TIMEOUT and the writer's REPLY_TIMEOUT.
  3. If failing stays the policy, have the error say that rank 0 queued more than BACKLOG jobs in one burst rather than "the peer is not keeping up".

Found on main at cc696a6b with DeepSeek V4.1 Flash at TP3 on three DGX Spark (GB10) hosts, after upgrading from a build without eager thinning and keeping its prefix-cache directory.

I have the fix in option 1 with the test above ready as a PR, if that approach works for you; happy to do option 2 instead if you prefer to keep thinning unbounded and make the link absorb bursts.

Lingua principale
Rust
Stelle
800
Fork
106
Merge medio
1g 3h
PR unite (30g)
100

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di NVIDIA/hyperflare-inference

Tutte le issue di NVIDIA/hyperflare-inference

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.