From 5ed8185d777fef7cc86f5178bd9c6442f010d776 Mon Sep 17 00:00:00 2001 From: chuan Date: Mon, 10 Aug 2026 10:59:09 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E5=A2=9E=E5=8A=A0=E6=96=B0=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E7=9B=B4=E6=8E=A5=E9=87=87=E6=A0=B7=E7=AD=96=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- TODOS.md | 9 +- dht-crawler/README.md | 4 +- dht-crawler/src/crawl_engine.rs | 12 ++ dht-crawler/src/lib.rs | 10 +- dht-crawler/src/peer_lookup.rs | 13 +- dht-crawler/src/runtime_stats.rs | 157 ++++++++++++++++++++- dht-crawler/src/sample_infohashes.rs | 200 +++++++++++++++++++++++++-- dht-crawler/src/scheduler.rs | 29 +++- dht-crawler/src/server.rs | 26 +++- dht-crawler/src/types.rs | 19 +++ dht-search.example.toml | 2 + dht-search/README.md | 4 + dht-search/src/api/handlers.rs | 16 +++ dht-search/src/api/response.rs | 16 +++ dht-search/src/config.rs | 21 +++ dht-search/src/monitor.rs | 11 ++ 16 files changed, 514 insertions(+), 35 deletions(-) diff --git a/TODOS.md b/TODOS.md index e6b1cb6..e141acb 100644 --- a/TODOS.md +++ b/TODOS.md @@ -255,7 +255,10 @@ - [x] 实现样本来源节点单点 `get_peers` 优先和失败后有限递归降级 - [x] 将 BEP-51 最大在途请求和采样失败后的迭代回退暴露为应用配置 - [x] 使用 Bitmagnet 等效并发完成一分钟资源测试并确认主网卡无丢包无错误且持久化无积压 -- [ ] 评估将新发现但尚未验证的节点直接作为 BEP-51 候选并避免与 `find_node` 重复探测 +- [x] 将新发现但尚未验证的节点按地址稳定分流到有界 BEP-51 通道并避免与 `find_node` 重复探测 +- [x] 增加直接采样候选队列请求响应重复过滤丢弃和 Metadata 成功来源转化指标 +- [x] 使用相同网络预算对旧快照采样和新节点直接采样完成十分钟对比 +- [x] 确认直接采样在相近 UDP 流量下 Metadata 成功数提高约百分之六点六且单条成功 UDP 成本降低约百分之六点六 - [x] 完成首轮三分钟对比并验证 Peer Lookup UDP 从 `278` 降至 `254` 且网络稳定 - [ ] 通过多轮或更长时间运行评估随机 DHT 样本下的 Metadata 成功率 - [ ] 统计按需验证的 Peer 发现率握手成功率和平均验证耗时 @@ -322,6 +325,6 @@ - [x] 将规模基准拆分为参数数据集工作负载采样报告和编排模块 - [x] 明确单元组件集成端到端和性能测试层级并增加公开 API 集成测试 -对比 Bitmagnet 新节点直接采样策略并测量有效样本数每次成功 Metadata 的 UDP 和 TCP 成本 +完成二十四小时持续运行并继续观察私有内存 Metadata 成功率候选队列深度和每条成功 Metadata 的网络成本 -随后完成二十四小时持续运行并根据数据决定资源参数和正则查询优化优先级 +随后增加磁盘剩余空间保护日志轮转和 RocksDB 检查点恢复 diff --git a/dht-crawler/README.md b/dht-crawler/README.md index 0552c3b..d60bcf6 100644 --- a/dht-crawler/README.md +++ b/dht-crawler/README.md @@ -180,7 +180,7 @@ let options = DHTOptions { | `DHTOptions` | 监听端口、网络模式、顶层队列和主动 UDP 查询总预算 | | `MetadataOptions` | 下载超时、队列、并发、每秒 TCP 建连和失败 Peer 缓存 | | `PeerLookupOptions` | 主动 `get_peers` 的速率与并发 | -| `SampleInfohashesOptions` | BEP-51 采样速率、并发、超时、退避和 Hash 去重容量 | +| `SampleInfohashesOptions` | BEP-51 采样速率、并发、新节点稳定分流、有界候选队列、超时、退避和 Hash 去重容量 | | `RateLimitOptions` | `find_node`、在途请求和 UDP 回复预算 | | `PoolOptions` | 节点池、最近探测记录和响应节点缓存 | | `BootstrapOptions` | Bootstrap 节点与失败退避 | @@ -191,6 +191,8 @@ let options = DHTOptions { - BEP-51 采样 hash 可以通过 `DHTServer::on_sampled_hashes` 批量异步准入 - 采样准入队列有固定容量并在压力升高时暂停新的 BEP-51 查询 +- `new_node_sample_percent` 大于零时按节点地址稳定分流并优先直接采样 避免同一次发现同时执行 `find_node` +- 新节点采样通道满时回退到抓取池 不会依靠无限队列维持吞吐 - 带首选节点的 Peer Lookup 先执行单点查询只有失败后才进入有限迭代查找 - `DHTOptions::default()` 使用 `Ipv4Only`； diff --git a/dht-crawler/src/crawl_engine.rs b/dht-crawler/src/crawl_engine.rs index 7d49f77..952e65b 100644 --- a/dht-crawler/src/crawl_engine.rs +++ b/dht-crawler/src/crawl_engine.rs @@ -13,6 +13,7 @@ use crate::routing_snapshot::RoutingSnapshot; #[cfg(test)] use crate::runtime_stats::DhtRuntimeLimits; use crate::runtime_stats::DhtRuntimeStats; +use crate::sample_infohashes::SampleCandidateRouter; use crate::types::{NetMode, NodeTuple}; use ahash::AHashMap; use arc_swap::ArcSwap; @@ -80,6 +81,7 @@ pub(crate) struct CrawlEngine { pub(crate) node_count: Arc, runtime_stats: DhtRuntimeStats, outbound_query_budget: SharedRateBudget, + sample_candidate_router: Option, } impl CrawlEngine { @@ -87,6 +89,7 @@ impl CrawlEngine { config: ResolvedCrawlConfig, runtime_stats: DhtRuntimeStats, outbound_query_budget: SharedRateBudget, + sample_candidate_router: Option, ) -> Self { let (priority_tx, priority_rx) = mpsc::channel(config.priority_event_channel_capacity); let (discovery_tx, discovery_rx) = mpsc::channel(config.discovery_event_channel_capacity); @@ -99,10 +102,18 @@ impl CrawlEngine { node_count: Arc::new(AtomicUsize::new(0)), runtime_stats, outbound_query_budget, + sample_candidate_router, } } pub(crate) fn route_discovered(&self, node: NodeTuple) { + if self + .sample_candidate_router + .as_ref() + .is_some_and(|router| router.route(node)) + { + return; + } let enqueue_result = self.discovery_tx.try_send(node); self.runtime_stats.set_crawl_discovery_queue_depth( self.discovery_tx @@ -935,6 +946,7 @@ mod tests { config, stats.clone(), SharedRateBudget::per_second(10_000, 10_000, true), + None, ); engine.route_discovered(node(1, "8.8.8.8:1")); diff --git a/dht-crawler/src/lib.rs b/dht-crawler/src/lib.rs index bf23708..7687877 100644 --- a/dht-crawler/src/lib.rs +++ b/dht-crawler/src/lib.rs @@ -40,7 +40,7 @@ pub use runtime_stats::{ pub use scheduler::{MetadataScheduler, MetadataSchedulerCallbacks, MetadataSchedulerLimits}; pub use server::{DHTServer, HashDiscovered}; pub use types::{ - BootstrapOptions, CrawlOptions, DHTOptions, FileInfo, MetadataFetchCompletion, + BootstrapOptions, CrawlOptions, DHTOptions, DiscoverySource, FileInfo, MetadataFetchCompletion, MetadataFetchCompletionStatus, MetadataOptions, NetMode, NodeTuple, PeerLookupOptions, PeerLookupResult, PoolOptions, RateLimitOptions, SampleInfohashesOptions, SchedulerOptions, TargetOptions, TorrentInfo, @@ -55,9 +55,9 @@ pub mod prelude { }; pub use crate::server::DHTServer; pub use crate::types::{ - BootstrapOptions, CrawlOptions, DHTOptions, FileInfo, MetadataFetchCompletion, - MetadataFetchCompletionStatus, MetadataOptions, NetMode, NodeTuple, PeerLookupOptions, - PeerLookupResult, PoolOptions, RateLimitOptions, SampleInfohashesOptions, SchedulerOptions, - TargetOptions, TorrentInfo, + BootstrapOptions, CrawlOptions, DHTOptions, DiscoverySource, FileInfo, + MetadataFetchCompletion, MetadataFetchCompletionStatus, MetadataOptions, NetMode, + NodeTuple, PeerLookupOptions, PeerLookupResult, PoolOptions, RateLimitOptions, + SampleInfohashesOptions, SchedulerOptions, TargetOptions, TorrentInfo, }; } diff --git a/dht-crawler/src/peer_lookup.rs b/dht-crawler/src/peer_lookup.rs index 5ee2952..e4e2287 100644 --- a/dht-crawler/src/peer_lookup.rs +++ b/dht-crawler/src/peer_lookup.rs @@ -7,7 +7,7 @@ use crate::protocol::DhtResponse; use crate::routing_snapshot::{RoutingSnapshot, xor_distance_cmp}; use crate::runtime_stats::DhtRuntimeStats; use crate::server::HashDiscovered; -use crate::types::{NetMode, NodeTuple, PeerLookupOptions, PeerLookupResult}; +use crate::types::{DiscoverySource, NetMode, NodeTuple, PeerLookupOptions, PeerLookupResult}; use ahash::{AHashMap, AHashSet}; use arc_swap::ArcSwap; use bytes::BytesMut; @@ -57,6 +57,7 @@ struct LookupState { deadline: Instant, preferred_phase: bool, allow_iterative_fallback: bool, + source: DiscoverySource, completion: Option>, } @@ -92,6 +93,7 @@ pub(crate) struct PeerLookupRequest { pub(crate) info_hash: [u8; 20], pub(crate) preferred_node: Option, pub(crate) allow_iterative_fallback: bool, + pub(crate) source: DiscoverySource, pub(crate) completion: Option>, } @@ -101,6 +103,7 @@ impl PeerLookupRequest { info_hash, preferred_node: None, allow_iterative_fallback: true, + source: DiscoverySource::ActiveLookup, completion: None, } } @@ -135,6 +138,7 @@ impl PeerLookupHandle { info_hash, preferred_node: None, allow_iterative_fallback: true, + source: DiscoverySource::ActiveLookup, completion: Some(completion), }) .await @@ -369,6 +373,7 @@ impl PeerLookupActor { deadline: now + LOOKUP_TIMEOUT, preferred_phase: preferred.is_some(), allow_iterative_fallback: request.allow_iterative_fallback, + source: request.source, completion: request.completion, }, ); @@ -495,12 +500,14 @@ impl PeerLookupActor { let lookup_id = pending.lookup_id; let preferred_succeeded = pending.preferred_phase && !discovered.is_empty(); let preferred_without_fallback = pending.preferred_phase && !state.allow_iterative_fallback; + let source = state.source; let _ = state; for peer in discovered { let event = HashDiscovered { info_hash: hash.clone(), peer_addr: peer, + source, discovered_at: now, }; if self.hash_tx.try_send(event).is_ok() { @@ -666,6 +673,7 @@ mod tests { addr: remote_addr, }), allow_iterative_fallback: true, + source: DiscoverySource::SampleDirect, completion: Some(completion), }) .await @@ -706,6 +714,7 @@ mod tests { .unwrap(); assert_eq!(event.info_hash, hex::encode([3; 20])); assert_eq!(event.peer_addr, "8.8.4.4:6881".parse().unwrap()); + assert_eq!(event.source, DiscoverySource::SampleDirect); let result = tokio::time::timeout(Duration::from_secs(1), result_rx) .await .unwrap() @@ -763,6 +772,7 @@ mod tests { addr: preferred_addr, }), allow_iterative_fallback: true, + source: DiscoverySource::SampleDirect, completion: None, }) .await @@ -848,6 +858,7 @@ mod tests { addr: preferred_addr, }), allow_iterative_fallback: false, + source: DiscoverySource::SampleDirect, completion: Some(completion), }) .await diff --git a/dht-crawler/src/runtime_stats.rs b/dht-crawler/src/runtime_stats.rs index 99a60c8..fc2499f 100644 --- a/dht-crawler/src/runtime_stats.rs +++ b/dht-crawler/src/runtime_stats.rs @@ -1,5 +1,6 @@ // 负责收集 DHT 抓取调度网络和 Metadata 运行指标 +use crate::types::DiscoverySource; use std::sync::Arc; use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering}; @@ -252,10 +253,26 @@ pub struct DhtRuntimeSnapshot { pub peer_lookup_preferred_succeeded: u64, /// Preferred-node attempts that fell back to iterative lookup. pub peer_lookup_fallbacks: u64, + /// Current newly discovered node sampling-lane depth. + pub sample_candidate_queue_depth: usize, + /// Configured newly discovered node sampling-lane capacity. + pub sample_candidate_queue_capacity: usize, + /// Newly discovered nodes routed directly to BEP-51 sampling. + pub sample_candidates_routed: u64, + /// Sampling candidates sent to the crawl pool because the bounded lane was full. + pub sample_candidates_fallback: u64, /// BEP-51 requests sent. pub sample_infohashes_queries: u64, + /// BEP-51 requests sent directly to newly discovered nodes. + pub sample_infohashes_direct_queries: u64, + /// BEP-51 requests sent to the responsive-node snapshot. + pub sample_infohashes_snapshot_queries: u64, /// Matched BEP-51 responses. pub sample_infohashes_responses: u64, + /// Matched responses from directly sampled new nodes. + pub sample_infohashes_direct_responses: u64, + /// Matched responses from responsive-node snapshot sampling. + pub sample_infohashes_snapshot_responses: u64, /// BEP-51 requests that timed out. pub sample_infohashes_timeouts: u64, /// BEP-51 UDP sends that failed immediately. @@ -274,6 +291,14 @@ pub struct DhtRuntimeSnapshot { pub metadata_peer_attempts: u64, /// Successful Peer downloads and parses. pub metadata_peer_succeeded: u64, + /// Successful Metadata whose winning Peer came from `announce_peer`. + pub metadata_success_from_announce: u64, + /// Successful Metadata whose winning Peer came from direct new-node sampling. + pub metadata_success_from_sample_direct: u64, + /// Successful Metadata whose winning Peer came from responsive snapshot sampling. + pub metadata_success_from_sample_snapshot: u64, + /// Successful Metadata whose winning Peer came from delayed active lookup. + pub metadata_success_from_active_lookup: u64, /// Failed real Peer attempts. pub metadata_peer_failed: u64, /// End-to-end Peer timeouts. @@ -366,8 +391,16 @@ struct DhtRuntimeStatsInner { peer_lookup_output_dropped: AtomicU64, peer_lookup_preferred_succeeded: AtomicU64, peer_lookup_fallbacks: AtomicU64, + sample_candidate_queue_depth: AtomicUsize, + sample_candidate_queue_capacity: AtomicUsize, + sample_candidates_routed: AtomicU64, + sample_candidates_fallback: AtomicU64, sample_infohashes_queries: AtomicU64, + sample_infohashes_direct_queries: AtomicU64, + sample_infohashes_snapshot_queries: AtomicU64, sample_infohashes_responses: AtomicU64, + sample_infohashes_direct_responses: AtomicU64, + sample_infohashes_snapshot_responses: AtomicU64, sample_infohashes_timeouts: AtomicU64, sample_infohashes_send_failures: AtomicU64, sample_infohashes_response_dropped: AtomicU64, @@ -377,6 +410,10 @@ struct DhtRuntimeStatsInner { sample_infohashes_hashes_dropped: AtomicU64, metadata_peer_attempts: AtomicU64, metadata_peer_succeeded: AtomicU64, + metadata_success_from_announce: AtomicU64, + metadata_success_from_sample_direct: AtomicU64, + metadata_success_from_sample_snapshot: AtomicU64, + metadata_success_from_active_lookup: AtomicU64, metadata_peer_failed: AtomicU64, metadata_peer_timeouts: AtomicU64, metadata_connect_failed: AtomicU64, @@ -479,8 +516,16 @@ impl Default for DhtRuntimeStatsInner { peer_lookup_output_dropped: AtomicU64::new(0), peer_lookup_preferred_succeeded: AtomicU64::new(0), peer_lookup_fallbacks: AtomicU64::new(0), + sample_candidate_queue_depth: AtomicUsize::new(0), + sample_candidate_queue_capacity: AtomicUsize::new(0), + sample_candidates_routed: AtomicU64::new(0), + sample_candidates_fallback: AtomicU64::new(0), sample_infohashes_queries: AtomicU64::new(0), + sample_infohashes_direct_queries: AtomicU64::new(0), + sample_infohashes_snapshot_queries: AtomicU64::new(0), sample_infohashes_responses: AtomicU64::new(0), + sample_infohashes_direct_responses: AtomicU64::new(0), + sample_infohashes_snapshot_responses: AtomicU64::new(0), sample_infohashes_timeouts: AtomicU64::new(0), sample_infohashes_send_failures: AtomicU64::new(0), sample_infohashes_response_dropped: AtomicU64::new(0), @@ -490,6 +535,10 @@ impl Default for DhtRuntimeStatsInner { sample_infohashes_hashes_dropped: AtomicU64::new(0), metadata_peer_attempts: AtomicU64::new(0), metadata_peer_succeeded: AtomicU64::new(0), + metadata_success_from_announce: AtomicU64::new(0), + metadata_success_from_sample_direct: AtomicU64::new(0), + metadata_success_from_sample_snapshot: AtomicU64::new(0), + metadata_success_from_active_lookup: AtomicU64::new(0), metadata_peer_failed: AtomicU64::new(0), metadata_peer_timeouts: AtomicU64::new(0), metadata_connect_failed: AtomicU64::new(0), @@ -619,8 +668,28 @@ impl DhtRuntimeStats { .peer_lookup_preferred_succeeded .load(Ordering::Relaxed), peer_lookup_fallbacks: inner.peer_lookup_fallbacks.load(Ordering::Relaxed), + sample_candidate_queue_depth: inner + .sample_candidate_queue_depth + .load(Ordering::Relaxed), + sample_candidate_queue_capacity: inner + .sample_candidate_queue_capacity + .load(Ordering::Relaxed), + sample_candidates_routed: inner.sample_candidates_routed.load(Ordering::Relaxed), + sample_candidates_fallback: inner.sample_candidates_fallback.load(Ordering::Relaxed), sample_infohashes_queries: inner.sample_infohashes_queries.load(Ordering::Relaxed), + sample_infohashes_direct_queries: inner + .sample_infohashes_direct_queries + .load(Ordering::Relaxed), + sample_infohashes_snapshot_queries: inner + .sample_infohashes_snapshot_queries + .load(Ordering::Relaxed), sample_infohashes_responses: inner.sample_infohashes_responses.load(Ordering::Relaxed), + sample_infohashes_direct_responses: inner + .sample_infohashes_direct_responses + .load(Ordering::Relaxed), + sample_infohashes_snapshot_responses: inner + .sample_infohashes_snapshot_responses + .load(Ordering::Relaxed), sample_infohashes_timeouts: inner.sample_infohashes_timeouts.load(Ordering::Relaxed), sample_infohashes_send_failures: inner .sample_infohashes_send_failures @@ -642,6 +711,18 @@ impl DhtRuntimeStats { .load(Ordering::Relaxed), metadata_peer_attempts: inner.metadata_peer_attempts.load(Ordering::Relaxed), metadata_peer_succeeded: inner.metadata_peer_succeeded.load(Ordering::Relaxed), + metadata_success_from_announce: inner + .metadata_success_from_announce + .load(Ordering::Relaxed), + metadata_success_from_sample_direct: inner + .metadata_success_from_sample_direct + .load(Ordering::Relaxed), + metadata_success_from_sample_snapshot: inner + .metadata_success_from_sample_snapshot + .load(Ordering::Relaxed), + metadata_success_from_active_lookup: inner + .metadata_success_from_active_lookup + .load(Ordering::Relaxed), metadata_peer_failed: inner.metadata_peer_failed.load(Ordering::Relaxed), metadata_peer_timeouts: inner.metadata_peer_timeouts.load(Ordering::Relaxed), metadata_connect_failed: inner.metadata_connect_failed.load(Ordering::Relaxed), @@ -890,16 +971,52 @@ impl DhtRuntimeStats { .fetch_add(1, Ordering::Relaxed); } - pub(crate) fn sample_query(&self) { + pub(crate) fn configure_sample_candidate_queue(&self, capacity: usize) { self.inner - .sample_infohashes_queries + .sample_candidate_queue_capacity + .store(capacity, Ordering::Relaxed); + } + + pub(crate) fn set_sample_candidate_queue_depth(&self, depth: usize) { + self.inner + .sample_candidate_queue_depth + .store(depth, Ordering::Relaxed); + } + + pub(crate) fn sample_candidate_routed(&self) { + self.inner + .sample_candidates_routed .fetch_add(1, Ordering::Relaxed); } - pub(crate) fn sample_response(&self) { + pub(crate) fn sample_candidate_fallback(&self) { + self.inner + .sample_candidates_fallback + .fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn sample_query(&self, direct: bool) { + self.inner + .sample_infohashes_queries + .fetch_add(1, Ordering::Relaxed); + let source = if direct { + &self.inner.sample_infohashes_direct_queries + } else { + &self.inner.sample_infohashes_snapshot_queries + }; + source.fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn sample_response(&self, direct: bool) { self.inner .sample_infohashes_responses .fetch_add(1, Ordering::Relaxed); + let source = if direct { + &self.inner.sample_infohashes_direct_responses + } else { + &self.inner.sample_infohashes_snapshot_responses + }; + source.fetch_add(1, Ordering::Relaxed); } pub(crate) fn sample_timeout(&self) { @@ -956,6 +1073,16 @@ impl DhtRuntimeStats { .fetch_add(1, Ordering::Relaxed); } + pub(crate) fn metadata_success_from(&self, source: DiscoverySource) { + let counter = match source { + DiscoverySource::AnnouncePeer => &self.inner.metadata_success_from_announce, + DiscoverySource::SampleDirect => &self.inner.metadata_success_from_sample_direct, + DiscoverySource::SampleSnapshot => &self.inner.metadata_success_from_sample_snapshot, + DiscoverySource::ActiveLookup => &self.inner.metadata_success_from_active_lookup, + }; + counter.fetch_add(1, Ordering::Relaxed); + } + pub(crate) fn metadata_peer_failed(&self) { self.inner .metadata_peer_failed @@ -1272,8 +1399,12 @@ mod tests { writer.peer_lookup_output_dropped(); writer.peer_lookup_preferred_succeeded(); writer.peer_lookup_fallback(); - writer.sample_query(); - writer.sample_response(); + writer.configure_sample_candidate_queue(20); + writer.set_sample_candidate_queue_depth(14); + writer.sample_candidate_routed(); + writer.sample_candidate_fallback(); + writer.sample_query(true); + writer.sample_response(true); writer.sample_timeout(); writer.sample_send_failed(); writer.sample_response_dropped(); @@ -1283,6 +1414,10 @@ mod tests { writer.sample_hash_dropped(); writer.metadata_peer_attempt(); writer.metadata_peer_succeeded(); + writer.metadata_success_from(DiscoverySource::AnnouncePeer); + writer.metadata_success_from(DiscoverySource::SampleDirect); + writer.metadata_success_from(DiscoverySource::SampleSnapshot); + writer.metadata_success_from(DiscoverySource::ActiveLookup); writer.metadata_peer_failed(); writer.metadata_peer_timeout(); writer.metadata_connect_failed(); @@ -1342,8 +1477,16 @@ mod tests { peer_lookup_output_dropped: 1, peer_lookup_preferred_succeeded: 1, peer_lookup_fallbacks: 1, + sample_candidate_queue_depth: 14, + sample_candidate_queue_capacity: 20, + sample_candidates_routed: 1, + sample_candidates_fallback: 1, sample_infohashes_queries: 1, + sample_infohashes_direct_queries: 1, + sample_infohashes_snapshot_queries: 0, sample_infohashes_responses: 1, + sample_infohashes_direct_responses: 1, + sample_infohashes_snapshot_responses: 0, sample_infohashes_timeouts: 1, sample_infohashes_send_failures: 1, sample_infohashes_response_dropped: 1, @@ -1353,6 +1496,10 @@ mod tests { sample_infohashes_hashes_dropped: 1, metadata_peer_attempts: 1, metadata_peer_succeeded: 1, + metadata_success_from_announce: 1, + metadata_success_from_sample_direct: 1, + metadata_success_from_sample_snapshot: 1, + metadata_success_from_active_lookup: 1, metadata_peer_failed: 1, metadata_peer_timeouts: 1, metadata_connect_failed: 1, diff --git a/dht-crawler/src/sample_infohashes.rs b/dht-crawler/src/sample_infohashes.rs index bac9ee7..423d59a 100644 --- a/dht-crawler/src/sample_infohashes.rs +++ b/dht-crawler/src/sample_infohashes.rs @@ -1,5 +1,6 @@ // 负责执行有界 BEP-51 采样并将准入后的 infohash 交给 Peer 查找 +use crate::addr::is_valid_node_addr; use crate::budget::{RateBucket, SharedRateBudget}; use crate::crawl_engine::CrawlEngine; use crate::krpc::{encode_sample_infohashes_query, for_each_response_node}; @@ -8,7 +9,7 @@ use crate::peer_lookup::{PeerLookupHandle, PeerLookupRequest}; use crate::protocol::DhtResponse; use crate::routing_snapshot::RoutingSnapshot; use crate::runtime_stats::DhtRuntimeStats; -use crate::types::{NetMode, NodeTuple, SampleInfohashesOptions}; +use crate::types::{DiscoverySource, NetMode, NodeTuple, SampleInfohashesOptions}; use ahash::{AHashMap, AHashSet}; use arc_swap::{ArcSwap, ArcSwapOption}; use bytes::BytesMut; @@ -39,6 +40,7 @@ struct PendingKey { #[derive(Debug, Clone, Copy)] struct PendingRequest { deadline: Instant, + source: DiscoverySource, } struct SampleResponse { @@ -49,6 +51,7 @@ struct SampleResponse { struct SampleAdmissionBatch { preferred_node: NodeTuple, + source: DiscoverySource, hashes: Vec<[u8; 20]>, } @@ -65,6 +68,73 @@ pub(crate) struct SampleInfohashesHandle { runtime_stats: DhtRuntimeStats, } +#[derive(Clone)] +pub(crate) struct SampleCandidateRouter { + sender: mpsc::Sender, + sample_percent: u8, + runtime_stats: DhtRuntimeStats, +} + +impl SampleCandidateRouter { + pub(crate) fn route(&self, node: NodeTuple) -> bool { + if self.sample_percent == 0 + || !is_valid_node_addr(&node.addr) + || stable_address_bucket(node.addr) >= self.sample_percent + { + return false; + } + match self.sender.try_send(node) { + Ok(()) => { + self.runtime_stats.sample_candidate_routed(); + self.runtime_stats.set_sample_candidate_queue_depth( + self.sender + .max_capacity() + .saturating_sub(self.sender.capacity()), + ); + true + } + Err(mpsc::error::TrySendError::Full(_)) => { + self.runtime_stats.sample_candidate_fallback(); + self.runtime_stats + .set_sample_candidate_queue_depth(self.sender.max_capacity()); + false + } + Err(mpsc::error::TrySendError::Closed(_)) => false, + } + } +} + +pub(crate) fn sample_candidate_lane( + options: &SampleInfohashesOptions, + runtime_stats: DhtRuntimeStats, +) -> (SampleCandidateRouter, mpsc::Receiver) { + let capacity = options.candidate_queue_capacity.max(1); + let (sender, receiver) = mpsc::channel(capacity); + runtime_stats.configure_sample_candidate_queue(capacity); + ( + SampleCandidateRouter { + sender, + sample_percent: options.new_node_sample_percent.min(100), + runtime_stats, + }, + receiver, + ) +} + +fn stable_address_bucket(addr: SocketAddr) -> u8 { + let mut hash = 0xcbf2_9ce4_8422_2325u64; + let mut mix = |byte: u8| { + hash ^= u64::from(byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + }; + match addr.ip() { + std::net::IpAddr::V4(ip) => ip.octets().into_iter().for_each(&mut mix), + std::net::IpAddr::V6(ip) => ip.octets().into_iter().for_each(&mut mix), + } + addr.port().to_be_bytes().into_iter().for_each(&mut mix); + (hash % 100) as u8 +} + impl SampleInfohashesHandle { pub(crate) fn route_response( &self, @@ -101,15 +171,27 @@ pub(crate) struct SampleInfohashesRuntime { pub(crate) shutdown: CancellationToken, } +pub(crate) struct SampleInfohashesInputs<'a> { + pub(crate) sockets: &'a std::collections::HashMap>, + pub(crate) snapshot: Arc>, + pub(crate) candidate_rx: mpsc::Receiver, + pub(crate) crawl_engine: Arc, + pub(crate) peer_lookup: PeerLookupHandle, +} + pub(crate) fn spawn_sample_infohashes( netmode: NetMode, local_id: [u8; 20], - sockets: &std::collections::HashMap>, - snapshot: Arc>, - crawl_engine: Arc, - peer_lookup: PeerLookupHandle, + inputs: SampleInfohashesInputs<'_>, runtime: SampleInfohashesRuntime, ) -> SampleInfohashesHandle { + let SampleInfohashesInputs { + sockets, + snapshot, + candidate_rx, + crawl_engine, + peer_lookup, + } = inputs; let SampleInfohashesRuntime { options, stats, @@ -140,6 +222,8 @@ pub(crate) fn spawn_sample_infohashes( socket_v4, socket_v6, snapshot, + candidate_rx, + direct_only: options.new_node_sample_percent > 0, crawl_engine, admission_tx, response_rx, @@ -177,6 +261,8 @@ struct SampleInfohashesActor { socket_v4: Option>, socket_v6: Option>, snapshot: Arc>, + candidate_rx: mpsc::Receiver, + direct_only: bool, crawl_engine: Arc, admission_tx: mpsc::Sender, response_rx: mpsc::Receiver, @@ -225,6 +311,7 @@ fn spawn_sample_admission( info_hash, preferred_node: Some(batch.preferred_node), allow_iterative_fallback: fallback_to_iterative, + source: batch.source, completion: None, }; let sent = tokio::select! { @@ -259,6 +346,7 @@ impl SampleInfohashesActor { } } } + self.runtime_stats.set_sample_candidate_queue_depth(0); } async fn dispatch(&mut self, now: Instant) { @@ -283,6 +371,23 @@ impl SampleInfohashesActor { .load() .random_nodes((budget * 8).max(64), filter_ipv6); let mut sent_count = 0usize; + while sent_count < budget { + let Some(node) = self.next_direct_candidate(now) else { + break; + }; + if self + .send_query(node, DiscoverySource::SampleDirect, now) + .await + { + sent_count += 1; + } + } + if self.direct_only { + if sent_count < budget { + self.query_budget.refund(budget - sent_count); + } + return; + } for node in candidates { if sent_count >= budget { break; @@ -295,7 +400,10 @@ impl SampleInfohashesActor { { continue; } - if self.send_query(node, now).await { + if self + .send_query(node, DiscoverySource::SampleSnapshot, now) + .await + { sent_count += 1; } } @@ -304,7 +412,25 @@ impl SampleInfohashesActor { } } - async fn send_query(&mut self, node: NodeTuple, now: Instant) -> bool { + fn next_direct_candidate(&mut self, now: Instant) -> Option { + for _ in 0..64 { + let node = self.candidate_rx.try_recv().ok()?; + self.runtime_stats + .set_sample_candidate_queue_depth(self.candidate_rx.len()); + if self.pending_addrs.contains(&node.addr) + || self + .next_allowed + .get(&node.addr) + .is_some_and(|deadline| *deadline > now) + { + continue; + } + return Some(node); + } + None + } + + async fn send_query(&mut self, node: NodeTuple, source: DiscoverySource, now: Instant) -> bool { let socket = if node.addr.is_ipv4() { self.socket_v4.clone() } else { @@ -331,11 +457,13 @@ impl SampleInfohashesActor { tid, }; let deadline = now + self.request_timeout; - self.pending.insert(key, PendingRequest { deadline }); + self.pending + .insert(key, PendingRequest { deadline, source }); self.pending_expiry.push_back((deadline, key)); self.pending_addrs.insert(node.addr); self.runtime_stats.udp_sent(buffer.len()); - self.runtime_stats.sample_query(); + self.runtime_stats + .sample_query(source == DiscoverySource::SampleDirect); #[cfg(feature = "metrics")] { counter!("dht_sample_infohashes_queries_total").increment(1); @@ -349,11 +477,12 @@ impl SampleInfohashesActor { addr: event.remote_addr, tid: event.tid, }; - if self.pending.remove(&key).is_none() { + let Some(pending) = self.pending.remove(&key) else { return; - } + }; self.pending_addrs.remove(&event.remote_addr); - self.runtime_stats.sample_response(); + self.runtime_stats + .sample_response(pending.source == DiscoverySource::SampleDirect); for_each_response_node(&event.response, self.netmode, |node| { self.crawl_engine.route_discovered(node) @@ -395,6 +524,7 @@ impl SampleInfohashesActor { .admission_tx .try_send(SampleAdmissionBatch { preferred_node, + source: pending.source, hashes, }) .is_ok() @@ -483,6 +613,16 @@ mod tests { use super::*; use crate::protocol::DhtMessage; + fn node_for_lane(want_direct: bool, percent: u8) -> NodeTuple { + (1..=u16::MAX) + .map(|port| NodeTuple { + id: [port as u8; 20], + addr: SocketAddr::from(([8, 8, 8, 8], port)), + }) + .find(|node| (stable_address_bucket(node.addr) < percent) == want_direct) + .expect("both routing lanes have an address") + } + #[test] fn sample_transaction_ids_have_a_reserved_tag() { assert!(is_sample_infohashes_tid(&[ @@ -508,6 +648,40 @@ mod tests { assert_eq!(response.samples.unwrap().len(), 40); } + #[test] + fn new_node_routing_is_stable_and_bounded() { + let options = SampleInfohashesOptions { + new_node_sample_percent: 50, + candidate_queue_capacity: 1, + ..SampleInfohashesOptions::default() + }; + let stats = DhtRuntimeStats::default(); + let (router, mut receiver) = sample_candidate_lane(&options, stats.clone()); + let direct = node_for_lane(true, 50); + let crawl = node_for_lane(false, 50); + + assert!(router.route(direct)); + assert!(!router.route(crawl)); + assert!(!router.route(direct)); + assert_eq!(receiver.try_recv().unwrap(), direct); + + let snapshot = stats.snapshot(); + assert_eq!(snapshot.sample_candidate_queue_capacity, 1); + assert_eq!(snapshot.sample_candidates_routed, 1); + assert_eq!(snapshot.sample_candidates_fallback, 1); + } + + #[test] + fn zero_percent_preserves_snapshot_only_sampling() { + let options = SampleInfohashesOptions { + new_node_sample_percent: 0, + ..SampleInfohashesOptions::default() + }; + let (router, mut receiver) = sample_candidate_lane(&options, DhtRuntimeStats::default()); + assert!(!router.route(node_for_lane(false, 0))); + assert!(receiver.try_recv().is_err()); + } + #[tokio::test] async fn batch_admission_only_forwards_application_approved_hashes() { let (batch_tx, batch_rx) = mpsc::channel(1); @@ -533,6 +707,7 @@ mod tests { id: [9; 20], addr: "127.0.0.1:6881".parse().unwrap(), }, + source: DiscoverySource::SampleDirect, hashes: vec![[1; 20], [2; 20]], }) .await @@ -544,6 +719,7 @@ mod tests { .unwrap(); assert_eq!(request.info_hash, [2; 20]); assert!(!request.allow_iterative_fallback); + assert_eq!(request.source, DiscoverySource::SampleDirect); let snapshot = stats.snapshot(); assert_eq!(snapshot.sample_infohashes_hashes_filtered, 1); assert_eq!(snapshot.sample_infohashes_hashes_discovered, 1); diff --git a/dht-crawler/src/scheduler.rs b/dht-crawler/src/scheduler.rs index 93592e7..d695487 100644 --- a/dht-crawler/src/scheduler.rs +++ b/dht-crawler/src/scheduler.rs @@ -6,7 +6,9 @@ use crate::peer_lookup::PeerLookupRequest; use crate::runtime_stats::DhtRuntimeLimits; use crate::runtime_stats::DhtRuntimeStats; use crate::server::HashDiscovered; -use crate::types::{MetadataFetchCompletion, MetadataFetchCompletionStatus, TorrentInfo}; +use crate::types::{ + DiscoverySource, MetadataFetchCompletion, MetadataFetchCompletionStatus, TorrentInfo, +}; use arc_swap::ArcSwapOption; #[cfg(feature = "metrics")] use metrics::{counter, gauge, histogram}; @@ -69,6 +71,7 @@ pub(crate) struct MetadataSchedulerRuntime { #[derive(Debug, Clone, Copy)] struct PeerCandidate { addr: SocketAddr, + source: DiscoverySource, discovered_at: Instant, } @@ -148,6 +151,7 @@ impl PendingHashQueue { entry.peers.retain(|peer| peer.addr != event.peer_addr); entry.peers.push(PeerCandidate { addr: event.peer_addr, + source: event.source, discovered_at, }); entry @@ -184,6 +188,7 @@ impl PendingHashQueue { info_hash, peers: vec![PeerCandidate { addr: event.peer_addr, + source: event.source, discovered_at: event.discovered_at, }], queued_at: now, @@ -450,6 +455,7 @@ impl MetadataScheduler { } else { let peer = PeerCandidate { addr: hash.peer_addr, + source: hash.source, discovered_at: hash.discovered_at, }; match state.peer_tx.try_send(peer) { @@ -690,6 +696,7 @@ impl MetadataScheduler { let attempts = attempts.load(Ordering::Relaxed); if let Some((peer, (name, total_size, files, piece_length))) = fetched { + runtime_stats.metadata_success_from(peer.source); let metadata = TorrentInfo { info_hash: job.info_hash.clone(), name, @@ -884,6 +891,7 @@ mod tests { HashDiscovered { info_hash: hash.to_string(), peer_addr: SocketAddr::from(([127, 0, 0, 1], port)), + source: DiscoverySource::AnnouncePeer, discovered_at, } } @@ -1061,14 +1069,17 @@ mod tests { let peers = vec![ PeerCandidate { addr: SocketAddr::from(([127, 0, 0, 1], 1000)), + source: DiscoverySource::AnnouncePeer, discovered_at: now, }, PeerCandidate { addr: SocketAddr::from(([127, 0, 0, 1], 1001)), + source: DiscoverySource::SampleDirect, discovered_at: now, }, PeerCandidate { addr: SocketAddr::from(([127, 0, 0, 1], 1002)), + source: DiscoverySource::ActiveLookup, discovered_at: now, }, ]; @@ -1134,10 +1145,12 @@ mod tests { let now = Instant::now(); let initial = vec![PeerCandidate { addr: SocketAddr::from(([127, 0, 0, 1], 1000)), + source: DiscoverySource::AnnouncePeer, discovered_at: now, }]; let live = PeerCandidate { addr: SocketAddr::from(([127, 0, 0, 1], 1001)), + source: DiscoverySource::ActiveLookup, discovered_at: now, }; let (peer_tx, mut peer_rx) = mpsc::channel(MAX_METADATA_PEERS_PER_HASH); @@ -1212,6 +1225,20 @@ mod tests { assert!(queue.contains("new")); } + #[test] + fn queue_preserves_peer_discovery_source() { + let start = Instant::now(); + let mut queue = PendingHashQueue::new(2, Duration::from_secs(60)); + let mut direct = event("direct", 1000, start); + direct.source = DiscoverySource::SampleDirect; + queue.push(direct, start); + + let entry = queue + .pop_newest_ready(start + Duration::from_secs(1), &HashMap::new()) + .unwrap(); + assert_eq!(entry.peers[0].source, DiscoverySource::SampleDirect); + } + #[test] fn queue_expires_stale_hashes_and_pops_newest_first() { let start = Instant::now(); diff --git a/dht-crawler/src/server.rs b/dht-crawler/src/server.rs index b26f0a2..9e59b57 100644 --- a/dht-crawler/src/server.rs +++ b/dht-crawler/src/server.rs @@ -17,15 +17,18 @@ use crate::response_limiter::{ }; use crate::runtime_stats::{DhtRuntimeLimits, DhtRuntimeStats}; use crate::sample_infohashes::{ - SampleHashAdmissionCallback, SampleInfohashesHandle, SampleInfohashesRuntime, - is_sample_infohashes_tid, spawn_sample_infohashes, + SampleHashAdmissionCallback, SampleInfohashesHandle, SampleInfohashesInputs, + SampleInfohashesRuntime, is_sample_infohashes_tid, sample_candidate_lane, + spawn_sample_infohashes, }; use crate::scheduler::{ MetadataCompletionCallback, MetadataFetchCallback, MetadataScheduler, MetadataSchedulerCallbacks, MetadataSchedulerLimits, MetadataSchedulerRuntime, TorrentAckCallback, }; -use crate::types::{DHTOptions, MetadataFetchCompletion, NetMode, NodeTuple, TorrentInfo}; +use crate::types::{ + DHTOptions, DiscoverySource, MetadataFetchCompletion, NetMode, NodeTuple, TorrentInfo, +}; use crate::udp_buffer::UdpBufferPool; use crate::udp_ingress::{WorkerHandle, spawn_udp_listener}; use arc_swap::ArcSwapOption; @@ -64,6 +67,8 @@ pub struct HashDiscovered { pub info_hash: String, /// Peer endpoint derived from announce `port`/`implied_port`. pub peer_addr: SocketAddr, + /// Network path that supplied this Peer candidate. + pub source: DiscoverySource, /// Monotonic discovery time used for freshness and queue ordering. pub discovered_at: std::time::Instant, } @@ -184,10 +189,13 @@ impl DHTServer { options.outbound_query_burst, true, ); + let (sample_candidate_router, sample_candidate_rx) = + sample_candidate_lane(&options.sample_infohashes, runtime_stats.clone()); let crawl_engine = Arc::new(CrawlEngine::new( crawl_config.clone(), runtime_stats.clone(), outbound_query_budget.clone(), + Some(sample_candidate_router), )); let peer_lookup = spawn_peer_lookup( options.netmode, @@ -205,10 +213,13 @@ impl DHTServer { let sample_infohashes = spawn_sample_infohashes( options.netmode, node_id, - &sockets_by_bind_addr, - crawl_engine.snapshot.clone(), - crawl_engine.clone(), - peer_lookup.clone(), + SampleInfohashesInputs { + sockets: &sockets_by_bind_addr, + snapshot: crawl_engine.snapshot.clone(), + candidate_rx: sample_candidate_rx, + crawl_engine: crawl_engine.clone(), + peer_lookup: peer_lookup.clone(), + }, SampleInfohashesRuntime { options: options.sample_infohashes.clone(), stats: runtime_stats.clone(), @@ -687,6 +698,7 @@ impl DHTServer { let event = HashDiscovered { info_hash: hash_hex, peer_addr: SocketAddr::new(addr.ip(), port), + source: DiscoverySource::AnnouncePeer, discovered_at: std::time::Instant::now(), }; diff --git a/dht-crawler/src/types.rs b/dht-crawler/src/types.rs index 10944c3..906c4b0 100644 --- a/dht-crawler/src/types.rs +++ b/dht-crawler/src/types.rs @@ -161,6 +161,19 @@ pub struct PeerLookupOptions { pub max_active_lookups: usize, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +/// Network path that supplied a Peer candidate for Metadata download. +pub enum DiscoverySource { + /// A remote node announced itself as a Peer. + AnnouncePeer, + /// A Peer was found for a Hash sampled directly from a newly discovered node. + SampleDirect, + /// A Peer was found for a Hash sampled from the responsive-node snapshot. + SampleSnapshot, + /// A Peer was found by the Metadata scheduler's delayed active lookup. + ActiveLookup, +} + #[derive(Debug, Clone, PartialEq, Eq)] /// Terminal result of one bounded active `get_peers` lookup. pub struct PeerLookupResult { @@ -179,6 +192,10 @@ pub struct SampleInfohashesOptions { pub burst: u32, /// Maximum outstanding BEP-51 requests. pub max_in_flight: usize, + /// Percentage of newly discovered node addresses routed directly to BEP-51 sampling. + pub new_node_sample_percent: u8, + /// Capacity of the bounded newly discovered node sampling lane. + pub candidate_queue_capacity: usize, /// Whether a sampled Hash should fall back to iterative get_peers after the sampling node /// returns no Peer. pub fallback_to_iterative: bool, @@ -336,6 +353,8 @@ impl Default for SampleInfohashesOptions { max_queries_per_second: 1, burst: 1, max_in_flight: 4, + new_node_sample_percent: 0, + candidate_queue_capacity: 4_096, fallback_to_iterative: true, request_timeout_millis: 1_500, unsupported_backoff_secs: 300, diff --git a/dht-search.example.toml b/dht-search.example.toml index d2255ea..c11ec09 100644 --- a/dht-search.example.toml +++ b/dht-search.example.toml @@ -25,6 +25,8 @@ metadata_workers = 400 metadata_connects_per_second = 400 sample_queries_per_second = 60 sample_max_in_flight = 100 +sample_new_node_percent = 50 +sample_candidate_queue_capacity = 8192 sample_fallback_to_iterative = false peer_lookups_per_second = 200 peer_lookup_max_active = 200 diff --git a/dht-search/README.md b/dht-search/README.md index 0abde32..a0998c6 100644 --- a/dht-search/README.md +++ b/dht-search/README.md @@ -51,6 +51,8 @@ cargo run -p dht-search -- --data-dir D:\data\dht-search --run-duration-secs 360 | `peer_lookup_max_active` | `200` | 同时运行的 Peer 查找数量 | | `sample_queries_per_second` | `60` | BEP-51 采样查询速率 | | `sample_max_in_flight` | `100` | 同时等待响应的 BEP-51 采样请求数量 | +| `sample_new_node_percent` | `50` | 按节点地址稳定分流到直接采样通道的比例 | +| `sample_candidate_queue_capacity` | `8192` | 新发现节点直接采样通道的有界容量 | | `sample_fallback_to_iterative` | `false` | 采样节点没有返回 Peer 时快速结束以扩大覆盖面 | | `metadata_workers` | `400` | 同时处理的 Metadata 任务数量 | | `metadata_connects_per_second` | `400` | 每秒真正开始的 Peer TCP 连接硬上限 | @@ -113,6 +115,8 @@ Invoke-RestMethod http://127.0.0.1:8080/stats | ConvertTo-Json -Depth 5 BEP-51 返回的 infohash 会先进入有界批量准入队列并由 RocksDB 精确判断 +新发现节点按地址稳定分流 同一地址只进入直接采样或 `find_node` 通道 直接采样队列满时会安全回退到抓取池 `/stats` 会分别报告候选队列直接采样请求响应 Hash 去重丢弃以及成功 Metadata 的网络来源 + 已有 infohash 只更新最后发现时间和发现次数不会再次执行 Peer Lookup 未知 infohash 首先只向返回样本的 DHT 节点查询一次 `get_peers` 只有单点查询没有返回 Peer 时才降级为有限迭代查找 diff --git a/dht-search/src/api/handlers.rs b/dht-search/src/api/handlers.rs index 75b6730..0583bd3 100644 --- a/dht-search/src/api/handlers.rs +++ b/dht-search/src/api/handlers.rs @@ -50,12 +50,28 @@ pub(crate) async fn stats(State(state): State) -> Json peer_lookup_queries: dht.peer_lookup_queries, peer_lookup_preferred_succeeded: dht.peer_lookup_preferred_succeeded, peer_lookup_fallbacks: dht.peer_lookup_fallbacks, + sample_candidate_queue: dht.sample_candidate_queue_depth, + sample_candidate_queue_capacity: dht.sample_candidate_queue_capacity, + sample_candidates_routed: dht.sample_candidates_routed, + sample_candidates_fallback: dht.sample_candidates_fallback, sample_queries: dht.sample_infohashes_queries, + sample_direct_queries: dht.sample_infohashes_direct_queries, + sample_snapshot_queries: dht.sample_infohashes_snapshot_queries, + sample_responses: dht.sample_infohashes_responses, + sample_direct_responses: dht.sample_infohashes_direct_responses, + sample_snapshot_responses: dht.sample_infohashes_snapshot_responses, + sample_timeouts: dht.sample_infohashes_timeouts, sampled_hashes: dht.sample_infohashes_hashes_discovered, sampled_hashes_filtered: dht.sample_infohashes_hashes_filtered, + sampled_hashes_duplicate: dht.sample_infohashes_hashes_duplicate, + sampled_hashes_dropped: dht.sample_infohashes_hashes_dropped, metadata_peer_attempts: dht.metadata_peer_attempts, metadata_in_flight: dht.metadata_in_flight, metadata_ok: dht.metadata_peer_succeeded, + metadata_ok_from_announce: dht.metadata_success_from_announce, + metadata_ok_from_sample_direct: dht.metadata_success_from_sample_direct, + metadata_ok_from_sample_snapshot: dht.metadata_success_from_sample_snapshot, + metadata_ok_from_active_lookup: dht.metadata_success_from_active_lookup, metadata_failed: dht.metadata_peer_failed, metadata_filtered: observability .metadata_failure_size_limit diff --git a/dht-search/src/api/response.rs b/dht-search/src/api/response.rs index 5aceee4..17c9305 100644 --- a/dht-search/src/api/response.rs +++ b/dht-search/src/api/response.rs @@ -22,12 +22,28 @@ pub(crate) struct StatsResponse { pub(crate) peer_lookup_queries: u64, pub(crate) peer_lookup_preferred_succeeded: u64, pub(crate) peer_lookup_fallbacks: u64, + pub(crate) sample_candidate_queue: usize, + pub(crate) sample_candidate_queue_capacity: usize, + pub(crate) sample_candidates_routed: u64, + pub(crate) sample_candidates_fallback: u64, pub(crate) sample_queries: u64, + pub(crate) sample_direct_queries: u64, + pub(crate) sample_snapshot_queries: u64, + pub(crate) sample_responses: u64, + pub(crate) sample_direct_responses: u64, + pub(crate) sample_snapshot_responses: u64, + pub(crate) sample_timeouts: u64, pub(crate) sampled_hashes: u64, pub(crate) sampled_hashes_filtered: u64, + pub(crate) sampled_hashes_duplicate: u64, + pub(crate) sampled_hashes_dropped: u64, pub(crate) metadata_peer_attempts: u64, pub(crate) metadata_in_flight: usize, pub(crate) metadata_ok: u64, + pub(crate) metadata_ok_from_announce: u64, + pub(crate) metadata_ok_from_sample_direct: u64, + pub(crate) metadata_ok_from_sample_snapshot: u64, + pub(crate) metadata_ok_from_active_lookup: u64, pub(crate) metadata_failed: u64, pub(crate) metadata_filtered: u64, pub(crate) metadata_filtered_too_large: u64, diff --git a/dht-search/src/config.rs b/dht-search/src/config.rs index 275e530..697b0bd 100644 --- a/dht-search/src/config.rs +++ b/dht-search/src/config.rs @@ -52,6 +52,8 @@ pub(crate) struct DhtConfig { pub(crate) metadata_connects_per_second: u32, pub(crate) sample_queries_per_second: u32, pub(crate) sample_max_in_flight: usize, + pub(crate) sample_new_node_percent: u8, + pub(crate) sample_candidate_queue_capacity: usize, pub(crate) sample_fallback_to_iterative: bool, pub(crate) peer_lookups_per_second: u32, pub(crate) peer_lookup_max_active: usize, @@ -155,6 +157,8 @@ impl AppConfig { max_queries_per_second: self.dht.sample_queries_per_second, burst: self.dht.sample_queries_per_second.max(1), max_in_flight: self.dht.sample_max_in_flight, + new_node_sample_percent: self.dht.sample_new_node_percent, + candidate_queue_capacity: self.dht.sample_candidate_queue_capacity, fallback_to_iterative: self.dht.sample_fallback_to_iterative, ..defaults.sample_infohashes }, @@ -241,11 +245,17 @@ impl AppConfig { || self.dht.outbound_query_burst == 0 || self.dht.metadata_connects_per_second == 0 || self.dht.sample_max_in_flight == 0 + || self.dht.sample_candidate_queue_capacity == 0 || self.dht.peer_lookup_max_active == 0 || self.dht.find_node_max_in_flight == 0 { return Err(AppError::Config("网络速率和并发上限必须大于零".to_owned())); } + if self.dht.sample_new_node_percent > 100 { + return Err(AppError::Config( + "sample_new_node_percent 必须在 0 到 100 之间".to_owned(), + )); + } Ok(()) } } @@ -293,6 +303,8 @@ impl Default for DhtConfig { metadata_connects_per_second: 400, sample_queries_per_second: 60, sample_max_in_flight: 100, + sample_new_node_percent: 50, + sample_candidate_queue_capacity: 8_192, sample_fallback_to_iterative: false, peer_lookups_per_second: 200, peer_lookup_max_active: 200, @@ -368,6 +380,8 @@ mod tests { assert_eq!(options.max_outbound_queries_per_second, 1_000); assert_eq!(options.sample_infohashes.max_queries_per_second, 60); assert_eq!(options.sample_infohashes.max_in_flight, 100); + assert_eq!(options.sample_infohashes.new_node_sample_percent, 50); + assert_eq!(options.sample_infohashes.candidate_queue_capacity, 8_192); assert!(!options.sample_infohashes.fallback_to_iterative); assert_eq!(options.peer_lookup.max_lookups_per_second, 200); assert_eq!(options.peer_lookup.max_active_lookups, 200); @@ -415,4 +429,11 @@ mod tests { config.metadata_limits.max_files = 0; assert!(matches!(config.validate(), Err(AppError::Config(_)))); } + + #[test] + fn invalid_new_node_sample_percent_is_rejected() { + let mut config = AppConfig::default(); + config.dht.sample_new_node_percent = 101; + assert!(matches!(config.validate(), Err(AppError::Config(_)))); + } } diff --git a/dht-search/src/monitor.rs b/dht-search/src/monitor.rs index 9ed30d6..30964a1 100644 --- a/dht-search/src/monitor.rs +++ b/dht-search/src/monitor.rs @@ -44,12 +44,23 @@ pub(crate) async fn run( peer_lookup_preferred_succeeded = dht.peer_lookup_preferred_succeeded, peer_lookup_fallbacks = dht.peer_lookup_fallbacks, sample_queries = dht.sample_infohashes_queries, + sample_direct_queries = dht.sample_infohashes_direct_queries, + sample_direct_responses = dht.sample_infohashes_direct_responses, + sample_candidate_queue = dht.sample_candidate_queue_depth, + sample_candidates_routed = dht.sample_candidates_routed, + sample_candidates_fallback = dht.sample_candidates_fallback, sampled_hashes = dht.sample_infohashes_hashes_discovered, sampled_hashes_filtered = dht.sample_infohashes_hashes_filtered, + sampled_hashes_duplicate = dht.sample_infohashes_hashes_duplicate, + sampled_hashes_dropped = dht.sample_infohashes_hashes_dropped, peers = dht.peer_lookup_peers_found, metadata_connects_per_second, metadata_in_flight = dht.metadata_in_flight, metadata_ok = dht.metadata_peer_succeeded, + metadata_ok_from_announce = dht.metadata_success_from_announce, + metadata_ok_from_sample_direct = dht.metadata_success_from_sample_direct, + metadata_ok_from_sample_snapshot = dht.metadata_success_from_sample_snapshot, + metadata_ok_from_active_lookup = dht.metadata_success_from_active_lookup, metadata_failed = dht.metadata_peer_failed, metadata_filtered = observability .metadata_failure_size_limit