From c356926d698581018d5cc876131eb6495da44a86 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=A1=A5=E4=B8=8B=E7=BA=A2=E8=8D=AF?= Date: Fri, 16 Jan 2026 19:46:51 +0800 Subject: [PATCH] Prepare for crates.io release v0.0.1 --- .gitignore | 17 +++++++++ Cargo.toml | 48 +++++++++++++------------- LICENSE | 21 ++++++++++++ {src => examples}/main.rs | 21 +++++------- src/lib.rs | 6 ++-- src/metadata.rs | 1 - src/scheduler.rs | 72 +++++++++++++++++++++++++++------------ src/server.rs | 47 +++---------------------- src/sharded.rs | 4 --- src/types.rs | 13 ++++--- 10 files changed, 136 insertions(+), 114 deletions(-) create mode 100644 .gitignore create mode 100644 LICENSE rename {src => examples}/main.rs (86%) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..03b6f9f --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# Rust +/target/ +Cargo.lock +**/*.rs.bk + +# IDE +.idea/ +.vscode/ +*.swp + +# 输出文件 +torrents/ +*.json +*.db + +# 日志 +*.log diff --git a/Cargo.toml b/Cargo.toml index 9e1ebef..aa1a895 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,43 +1,46 @@ [package] -name = "dht_crawler" -version = "3.0.0" +name = "dht-crawler" +version = "0.0.1" edition = "2021" +authors = ["桥下红药 <1121744186@qq.com>"] +description = "高性能的 Rust DHT（分布式哈希表）爬虫库，用于爬取 BitTorrent DHT 网络中的种子信息" +license = "MIT" +documentation = "https://docs.rs/dht-crawler" +keywords = ["dht", "bittorrent", "crawler", "p2p", "torrent", "metadata"] +categories = ["network-programming", "asynchronous"] +readme = "README.md" + +[lib] +name = "dht_crawler" +path = "src/lib.rs" [dependencies] tokio = { version = "1.35", features = ["full"] } serde = { version = "1.0", features = ["derive"] } -serde_bencode = "0.2" # 更成熟的 bencode 库，支持 UTF-8 +serde_bencode = "0.2" sha1 = "0.10" hex = "0.4" rand = "0.8" - log = "0.4" -tracing = "0.1.43" -tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } -tracing-appender = "0.2" -tracing-log = "0.2" - - thiserror = "1.0" - socket2 = { version = "0.5", features = ["all"] } rbit = "0.1" bytes = "1.0" - -mimalloc = "0.1" bloomfilter = "1.0" -ahash = "0.8" # 快速哈希算法（比 SHA1 快 10倍+） - -# 可选：用于 Web API -actix-web = { version = "4.4", optional = true } -encoding_rs = { version = "0.8.35", optional = true } +ahash = "0.8" serde_bytes = "0.11.19" +mimalloc = { version = "0.1", optional = true } +[dev-dependencies] +tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } [features] default = [] -web = ["actix-web"] -encoding = ["dep:encoding_rs"] +mimalloc = ["dep:mimalloc"] + +[[example]] +name = "dht_crawler_example" +path = "examples/main.rs" # ==================== 性能优化配置 ==================== @@ -52,8 +55,3 @@ strip = true # 移除调试符号 - 减小二进制 [profile.dev] opt-level = 0 debug = true - -# 性能测试配置 -[profile.bench] -inherits = "release" -debug = true # 保留符号以便性能分析 diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..ecb2e4a --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2024 dht-crawler contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/src/main.rs b/examples/main.rs similarity index 86% rename from src/main.rs rename to examples/main.rs index 55c2348..eccb1fa 100644 --- a/src/main.rs +++ b/examples/main.rs @@ -1,9 +1,9 @@ +#[cfg(feature = "mimalloc")] #[global_allocator] -static GLOBAL: MiMalloc = MiMalloc; +static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; use dht_crawler::prelude::*; use std::sync::Arc; -use mimalloc::MiMalloc; use tracing_subscriber::EnvFilter; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -26,7 +26,8 @@ async fn main() -> Result<()> { metadata_timeout: 3, // ✅ 快速超时，快速失败 max_metadata_queue_size: 100000, // ✅ 大缓冲区（防止饱和） max_metadata_worker_count: 1000, // ✅ 激进并发（最大化吞吐） - netmode: NetMode::DualStack, // 网络模式：Ipv4Only（仅IPv4）、Ipv6Only（仅IPv6）、DualStack（双栈，默认） + netmode: NetMode::Ipv4Only, // 网络模式：Ipv4Only（仅IPv4）、Ipv6Only（仅IPv6）、DualStack（双栈，默认） + ..Default::default() // 使用默认值填充其他字段（节点队列容量等） }; // 统计计数器 @@ -72,12 +73,7 @@ async fn main() -> Result<()> { true }); - server.on_duplicate(|_hash| { - - }); - // 启动监控任务 - let dht_monitor = server.clone(); let count_monitor = torrent_count.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); @@ -88,10 +84,10 @@ async fn main() -> Result<()> { let success_fetch = count_monitor.load(Ordering::Relaxed); let uptime = start_time.elapsed().as_secs(); - // ✅ 监控：布隆过滤器的位使用情况反映了爬虫的活跃度 + // ✅ 监控：爬虫运行状态 log::info!( - "📊 [监控] 时长: {}s | 成功抓取: ✨ {} | 活跃指纹: {}", - uptime, success_fetch, dht_monitor.get_seen_count() + "📊 [监控] 时长: {}s | 成功抓取: ✨ {}", + uptime, success_fetch ); if uptime > 0 && success_fetch > 0 { @@ -103,4 +99,5 @@ async fn main() -> Result<()> { server.start().await?; Ok(()) -} \ No newline at end of file +} + diff --git a/src/lib.rs b/src/lib.rs index 16f907a..7747cd1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,9 +2,9 @@ mod error; mod server; pub mod protocol; pub mod types; -pub mod metadata; // 公开 metadata 模块 -mod sharded; // 分片锁模块 -pub mod scheduler; // 元数据调度器 +pub mod metadata; +mod sharded; +pub mod scheduler; pub use error::{DHTError, Result}; pub use server::{DHTServer, HashDiscovered}; diff --git a/src/metadata.rs b/src/metadata.rs index 8023ea7..40e9b9f 100644 --- a/src/metadata.rs +++ b/src/metadata.rs @@ -33,7 +33,6 @@ impl RbitFetcher { let peer_id = PeerId::generate(); - // 🔥 修改点：缩短连接超时到 3 秒 // DHT 网络很不稳定，如果 3 秒连不上，基本就是连不上了，不要浪费时间 let mut conn = match timeout( Duration::from_secs(3), diff --git a/src/scheduler.rs b/src/scheduler.rs index df0a200..40ede20 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -4,6 +4,7 @@ use crate::metadata::RbitFetcher; use std::sync::{Arc, RwLock}; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use tokio::sync::{mpsc, Mutex}; +#[cfg(debug_assertions)] use std::time::Duration; type TorrentCallback = Arc; @@ -124,37 +125,63 @@ impl MetadataScheduler { } // 主循环：只负责接收 hash 并转发到 worker 队列 + #[cfg(debug_assertions)] let mut stats_interval = tokio::time::interval(Duration::from_secs(60)); + #[cfg(debug_assertions)] stats_interval.tick().await; loop { - tokio::select! { - Some(hash) = self.hash_rx.recv() => { - self.total_received.fetch_add(1, Ordering::Relaxed); - - // 尝试发送到 worker 队列 - match task_tx.try_send(hash) { - Ok(_) => { - // 成功入队，增加计数器 - self.queue_len.fetch_add(1, Ordering::Relaxed); + #[cfg(debug_assertions)] + { + tokio::select! { + Some(hash) = self.hash_rx.recv() => { + self.total_received.fetch_add(1, Ordering::Relaxed); + + // 尝试发送到 worker 队列 + match task_tx.try_send(hash) { + Ok(_) => { + // 成功入队，增加计数器 + self.queue_len.fetch_add(1, Ordering::Relaxed); + } + Err(mpsc::error::TrySendError::Full(_)) => { + // 队列满，丢弃 + self.total_dropped.fetch_add(1, Ordering::Relaxed); + } + Err(_) => break, // Channel 关闭 } - Err(mpsc::error::TrySendError::Full(_)) => { - // 队列满，丢弃 - self.total_dropped.fetch_add(1, Ordering::Relaxed); - } - Err(_) => break, // Channel 关闭 } + + _ = stats_interval.tick() => { + self.print_stats(&task_tx); + } + + else => break, } - - _ = stats_interval.tick() => { - self.print_stats(&task_tx); + } + + #[cfg(not(debug_assertions))] + { + match self.hash_rx.recv().await { + Some(hash) => { + self.total_received.fetch_add(1, Ordering::Relaxed); + + // 尝试发送到 worker 队列 + match task_tx.try_send(hash) { + Ok(_) => { + // 成功入队，增加计数器 + self.queue_len.fetch_add(1, Ordering::Relaxed); + } + Err(mpsc::error::TrySendError::Full(_)) => { + // 队列满，丢弃 + self.total_dropped.fetch_add(1, Ordering::Relaxed); + } + Err(_) => break, // Channel 关闭 + } + } + None => break, // Channel 关闭 } - - else => break, } } - - // log::info!("🛑 Metadata 调度器停止"); } /// 处理单个 hash（Worker 调用） @@ -221,7 +248,8 @@ impl MetadataScheduler { } } - /// 输出统计信息 + /// 输出统计信息（仅在 debug 模式下编译） + #[cfg(debug_assertions)] fn print_stats(&self, task_tx: &mpsc::Sender) { let received = self.total_received.load(Ordering::Relaxed); let dropped = self.total_dropped.load(Ordering::Relaxed); diff --git a/src/server.rs b/src/server.rs index 90a1367..549b074 100644 --- a/src/server.rs +++ b/src/server.rs @@ -3,7 +3,7 @@ use crate::metadata::RbitFetcher; use crate::protocol::{DhtMessage, DhtArgs, DhtResponse}; use crate::scheduler::MetadataScheduler; use crate::types::{DHTOptions, TorrentInfo, NetMode}; -use crate::sharded::{ShardedBloom, ShardedNodeQueue, NodeTuple}; +use crate::sharded::{ShardedNodeQueue, NodeTuple}; use rand::Rng; use ahash::AHasher; use std::hash::{Hash, Hasher}; @@ -40,7 +40,6 @@ pub struct HashDiscovered { type TorrentCallback = Arc; type FilterCallback = Arc bool + Send + Sync>; -type DuplicateCallback = Arc; #[derive(Clone)] pub struct DHTServer { @@ -53,12 +52,10 @@ pub struct DHTServer { callback: Arc>>, filter: Arc>>, - on_duplicate: Arc>>, on_metadata_fetch: Arc>>, // 使用分片锁，大幅减少竞争 node_queue: Arc, - seen_hashes: Arc, // 发送 hash 发现事件 hash_tx: mpsc::Sender, @@ -138,13 +135,9 @@ impl DHTServer { let mut rng = rand::thread_rng(); let token_secret: Vec = (0..10).map(|_| rng.gen()).collect(); - // 使用分片队列和分片布隆过滤器 - // 队列容量：100000 个节点（扩容以适应 DHT 网络裂变速度） - let node_queue = ShardedNodeQueue::new(100000); - - // 布隆过滤器：预期500万元素，0.1%误判率 - // 内存使用：约 90MB（32分片 × 2.8MB） - let bloom = ShardedBloom::new_for_fp_rate(5_000_000, 0.001); + // 使用分片队列 + // 队列容量：从配置获取 + let node_queue = ShardedNodeQueue::new(options.node_queue_capacity); // ----------------------------------------------------------- // 内部初始化 MetadataScheduler @@ -184,9 +177,7 @@ impl DHTServer { callback, on_metadata_fetch, node_queue: Arc::new(node_queue), - seen_hashes: Arc::new(bloom), filter: Arc::new(RwLock::new(None)), - on_duplicate: Arc::new(RwLock::new(None)), hash_tx, metadata_queue_len, max_metadata_queue_size: options.max_metadata_queue_size, @@ -302,22 +293,6 @@ impl DHTServer { *self.filter.write().unwrap() = Some(Arc::new(filter)); } - /// 设置重复 Hash 发现的回调 - /// - /// 当接收到的 Hash 已经被布隆过滤器标记为“已存在”时调用。 - /// - /// # 注意事项 - /// - 库内部已经自动为每次调用包裹了 `tokio::spawn`。 - /// - 因此你可以放心地在回调中执行耗时操作（如数据库记录），而不用担心阻塞 UDP 线程。 - /// - 虽然内部有 spawn，但频繁触发仍会产生大量任务，请注意资源控制。 - pub fn on_duplicate(&self, callback: F) where F: Fn(&str) + Send + Sync + 'static { - *self.on_duplicate.write().unwrap() = Some(Arc::new(callback)); - } - - pub fn get_seen_count(&self) -> usize { - // 分片布隆过滤器的位数统计 - self.seen_hashes.number_of_bits() as usize - } pub fn get_node_pool_size(&self) -> usize { self.node_queue.len() @@ -566,20 +541,6 @@ impl DHTServer { }; let hash_hex = hex::encode(info_hash_arr); - // 使用分片布隆过滤器进行高效去重 - let is_duplicate = self.seen_hashes.check_and_set(&info_hash_arr); - - if is_duplicate { - let dup_cb = self.on_duplicate.read().unwrap().clone(); - if let Some(cb) = dup_cb { - let hash_hex_clone = hash_hex.clone(); - tokio::spawn(async move { - cb(&hash_hex_clone); - }); - } - return Ok(()); - } - let filter_cb = self.filter.read().unwrap().clone(); if let Some(f) = filter_cb { if !f(&hash_hex) { return Ok(()); } diff --git a/src/sharded.rs b/src/sharded.rs index e7f80e7..8409583 100644 --- a/src/sharded.rs +++ b/src/sharded.rs @@ -1,8 +1,6 @@ // 分片锁实现 - 大幅减少锁竞争，提升并发性能 // // 核心思想：1个大锁 → N个小锁 -// 性能提升：预期 3-4 倍 - use bloomfilter::Bloom; use std::collections::{HashSet, VecDeque}; use std::net::SocketAddr; @@ -226,8 +224,6 @@ impl ShardedNodeQueue { } /// 获取随机节点（用于DHT响应） - /// 🚀 优化：IPv4 和 IPv6 分开存储，直接从对应队列获取，无需过滤 - /// /// # Arguments /// * `count` - 需要获取的节点数量 /// * `filter_ipv6` - 如果为 `Some(true)`，只返回 IPv6 节点；如果为 `Some(false)`，只返回 IPv4 节点；如果为 `None`，返回所有节点（混合） diff --git a/src/types.rs b/src/types.rs index 08cc9f6..074fa4c 100644 --- a/src/types.rs +++ b/src/types.rs @@ -82,21 +82,26 @@ pub struct DHTOptions { /// 网络模式配置（仅IPv4、仅IPv6、或双栈） pub netmode: NetMode, + + /// 节点队列容量（默认 100000） + pub node_queue_capacity: usize, } impl Default for DHTOptions { fn default() -> Self { Self { - port: 0, + port: 6881, // BitTorrent DHT 默认端口 auto_metadata: true, // 缩短超时，快速失败，不等待慢节点 - metadata_timeout: 10, + metadata_timeout: 3, // 加大队列，防止流量高峰丢包 - max_metadata_queue_size: 10000, + max_metadata_queue_size: 100000, // 提高并发，模拟 Node.js 的高并发 IO max_metadata_worker_count: 1000, // 默认双栈 - netmode: NetMode::DualStack, + netmode: NetMode::Ipv4Only, + // 节点队列容量：100000 个节点（扩容以适应 DHT 网络裂变速度） + node_queue_capacity: 100000, } } } \ No newline at end of file