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

オープン
  • 何か問題がありますか？
#189 コメント 1 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

まだ誰も着手していません。

評価

難易度
3/5
見積もり時間
1〜2日
初心者へのやさしさ
42/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
rust
領域
backend, performance

調査の方向性

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.

索引モデルが issue の本文から書いたものです。

説明

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.

主要言語
Rust
スター
800
フォーク
106
平均マージ
1日 3時間
マージ済み PR（30日）
100

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

NVIDIA/hyperflare-inference のほかの issue

NVIDIA/hyperflare-inference の issue をすべて見る

似ている issue

Rust の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。