From 621a87557cde620e019c6df7abae32007ab2825d Mon Sep 17 00:00:00 2001 From: hailong15 Date: Thu, 3 Nov 2022 17:51:57 +0800 Subject: [PATCH 01/24] shards test --- tests/src/shard_test.rs | 21 +++++++++++++++++++++ tests_integration/src/redis/basic/mod.rs | 1 + tests_integration/src/redis/basic/shard.rs | 16 ++++++++++++++++ 3 files changed, 38 insertions(+) create mode 100644 tests_integration/src/redis/basic/shard.rs diff --git a/tests/src/shard_test.rs b/tests/src/shard_test.rs index 87dc29b81..ce432e213 100644 --- a/tests/src/shard_test.rs +++ b/tests/src/shard_test.rs @@ -1,6 +1,7 @@ //#![feature(map_first_last)] use std::collections::BTreeMap; +use std::fmt::format; use std::{ fs::File, io::{BufRead, BufReader}, @@ -165,6 +166,26 @@ fn shards_check() { md5(&key); } +// #[test] +// fn print_shards_check() { +// let shard_count = 4; +// let mut servers = Vec::with_capacity(shard_count); +// for i in 0..shard_count { +// servers.push(format!("192.168.0.{}", i).to_string()); +// } +// let hasher = Hasher::from("crc32local"); +// let dist = Distribute::from("modula", &servers); + +// for i in 1..=20 { +// let key = format!("test_shards_{}", i); +// println!( +// "{}: shards {}", +// key, +// dist.index(hasher.hash(&key.as_bytes())) +// ); +// } +// } + fn shard_check_with_files(path: String, hasher: &Hasher, dist: &Distribute) { shard_check_short_with_files(path.clone(), hasher, dist); diff --git a/tests_integration/src/redis/basic/mod.rs b/tests_integration/src/redis/basic/mod.rs index f2a644d63..2b3fb87fe 100644 --- a/tests_integration/src/redis/basic/mod.rs +++ b/tests_integration/src/redis/basic/mod.rs @@ -1,3 +1,4 @@ mod collection; mod conn; mod key; +mod shard; diff --git a/tests_integration/src/redis/basic/shard.rs b/tests_integration/src/redis/basic/shard.rs new file mode 100644 index 000000000..2324f616a --- /dev/null +++ b/tests_integration/src/redis/basic/shard.rs @@ -0,0 +1,16 @@ +//! 需要先指定分片，多条命令配合的测试 + +use crate::ci::env::*; +use crate::redis::RESTYPE; +use crate::redis_helper::*; +use function_name::named; +use redis::{Commands, RedisError}; +use std::collections::HashSet; +use std::vec; + +const SERVERS: [[&str; 2]; 4] = [ + ["127.0.0.1:56378", "127.0.0.1:56381"], + ["127.0.0.1:56378", "127.0.0.1:56380"], + ["127.0.0.1:56380", "127.0.0.1:56379"], + ["127.0.0.1:56381", "127.0.0.1:56378"], +]; From 3af8a4179304bdfb212c5f26d1389a899958bb70 Mon Sep 17 00:00:00 2001 From: hailong15 Date: Thu, 3 Nov 2022 17:51:57 +0800 Subject: [PATCH 02/24] shards test --- tests/src/shard_test.rs | 21 +++++++++ tests_integration/src/redis/basic/mod.rs | 1 + tests_integration/src/redis/basic/shard.rs | 55 ++++++++++++++++++++++ tests_integration/src/redis_helper.rs | 2 +- 4 files changed, 78 insertions(+), 1 deletion(-) create mode 100644 tests_integration/src/redis/basic/shard.rs diff --git a/tests/src/shard_test.rs b/tests/src/shard_test.rs index 87dc29b81..ce432e213 100644 --- a/tests/src/shard_test.rs +++ b/tests/src/shard_test.rs @@ -1,6 +1,7 @@ //#![feature(map_first_last)] use std::collections::BTreeMap; +use std::fmt::format; use std::{ fs::File, io::{BufRead, BufReader}, @@ -165,6 +166,26 @@ fn shards_check() { md5(&key); } +// #[test] +// fn print_shards_check() { +// let shard_count = 4; +// let mut servers = Vec::with_capacity(shard_count); +// for i in 0..shard_count { +// servers.push(format!("192.168.0.{}", i).to_string()); +// } +// let hasher = Hasher::from("crc32local"); +// let dist = Distribute::from("modula", &servers); + +// for i in 1..=20 { +// let key = format!("test_shards_{}", i); +// println!( +// "{}: shards {}", +// key, +// dist.index(hasher.hash(&key.as_bytes())) +// ); +// } +// } + fn shard_check_with_files(path: String, hasher: &Hasher, dist: &Distribute) { shard_check_short_with_files(path.clone(), hasher, dist); diff --git a/tests_integration/src/redis/basic/mod.rs b/tests_integration/src/redis/basic/mod.rs index f2a644d63..2b3fb87fe 100644 --- a/tests_integration/src/redis/basic/mod.rs +++ b/tests_integration/src/redis/basic/mod.rs @@ -1,3 +1,4 @@ mod collection; mod conn; mod key; +mod shard; diff --git a/tests_integration/src/redis/basic/shard.rs b/tests_integration/src/redis/basic/shard.rs new file mode 100644 index 000000000..b234833b4 --- /dev/null +++ b/tests_integration/src/redis/basic/shard.rs @@ -0,0 +1,55 @@ +//! 需要先指定分片，多条命令配合的测试 + +use crate::ci::env::*; +use crate::redis::RESTYPE; +use crate::redis_helper::*; +use function_name::named; +use redis::{Commands, RedisError}; +use std::collections::HashSet; +use std::vec; + +// const SERVERS: [[&str; 2]; 4] = [ +// ["127.0.0.1:56378", "127.0.0.1:56378"], +// ["127.0.0.1:56379", "127.0.0.1:56379"], +// ["127.0.0.1:56380", "127.0.0.1:56380"], +// ["127.0.0.1:56381", "127.0.0.1:56381"], +// ]; +const SERVERS: [[&str; 2]; 4] = [ + ["10.182.27.228:56378", "10.182.27.228:56378"], + ["10.182.27.228:56379", "10.182.27.228:56379"], + ["10.182.27.228:56380", "10.182.27.228:56380"], + ["10.182.27.228:56381", "10.182.27.228:56381"], +]; + +// crc32local % 4 的分片 +// test_shards_1: shards 2 +// test_shards_2: shards 0 +// test_shards_3: shards 2 +// test_shards_4: shards 3 +// test_shards_5: shards 1 +// test_shards_6: shards 1 +// test_shards_7: shards 3 +// test_shards_8: shards 2 +// test_shards_9: shards 0 +// test_shards_10: shards 1 +// test_shards_11: shards 3 +// test_shards_12: shards 3 +// test_shards_13: shards 1 +// test_shards_14: shards 0 +// test_shards_15: shards 2 +// test_shards_16: shards 0 +// test_shards_17: shards 2 +// test_shards_18: shards 3 +// test_shards_19: shards 1 +// test_shards_20: shards 2 + +/// hashrandomq, master + hashkeyq +/// hashrandomq 通过mesh set，然后直接连接后端读取，其分片应该是随机的 +/// 读取100次，每个分片>5, 则测试通过 +#[test] +fn test_hashrandomq() { + let mut con = get_conn(&RESTYPE.get_host()); + con.send_packed_command(&redis::cmd("master").get_packed_command()) + .expect("send master err"); + assert_eq!(con.set("a", 1), Ok(true)); +} diff --git a/tests_integration/src/redis_helper.rs b/tests_integration/src/redis_helper.rs index 2d5e69466..386d12620 100644 --- a/tests_integration/src/redis_helper.rs +++ b/tests_integration/src/redis_helper.rs @@ -1,5 +1,5 @@ +use redis::Client; pub fn get_conn(host: &str) -> redis::Connection { - use redis::Client; let host = String::from("redis://") + host; let client = Client::open(host).expect("get client err"); client.get_connection().expect("get conn err") From d4fc0d7997d802ccfd449759ae56d09163e6396a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=B3=A2?= Date: Fri, 4 Nov 2022 10:08:00 +0800 Subject: [PATCH 03/24] =?UTF-8?q?fix=20bug=EF=BC=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- endpoint/src/msgque/strategy/hitfirst.rs | 32 +++++++++++++----------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/endpoint/src/msgque/strategy/hitfirst.rs b/endpoint/src/msgque/strategy/hitfirst.rs index 6eef14b3d..567cac9ab 100644 --- a/endpoint/src/msgque/strategy/hitfirst.rs +++ b/endpoint/src/msgque/strategy/hitfirst.rs @@ -106,30 +106,32 @@ impl HitFirstReader { // 获取下一个读取node id let next = self.cursor_current.fetch_add(1, Ordering::Relaxed); let cursor_idx = next % self.cursors.len(); - let chits = self.cursors.get(cursor_idx).unwrap(); - let mut idx = chits.node_idx.load(Ordering::Relaxed); - let hits = chits.hits.fetch_add(1, Ordering::Relaxed); + let cursor = self.cursors.get(cursor_idx).unwrap(); + let mut node_idx = cursor.node_idx.load(Ordering::Relaxed); + let node_hits = cursor.hits.fetch_add(1, Ordering::Relaxed); // 如果当前位置hits数太大，也偏移一个位置 - if hits > MAX_HITS { - if let Ok(_c) = - chits - .node_idx - .compare_exchange(idx, idx + 1, Ordering::Relaxed, Ordering::Relaxed) - { - chits.hits.store(0, Ordering::Relaxed); + if node_hits > MAX_HITS { + let next_idx = (node_idx + 1) % self.qnodes.len(); + if let Ok(_c) = cursor.node_idx.compare_exchange( + node_idx, + next_idx, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + cursor.hits.store(0, Ordering::Relaxed); } - // cursor已偏移到新位置 - idx += 1; + // cursor已偏移到新位置，性能 + node_idx = next_idx; } assert!( - idx < self.qnodes.len(), + node_idx < self.qnodes.len(), "idx:{}, qunodes:{:?}", - idx, + node_idx, self.qnodes ); - let node = self.qnodes.get(idx).unwrap(); + let node = self.qnodes.get(node_idx).unwrap(); log::debug!("+++ use common cursor:{}, {:?}", cursor_idx, self); node.id } From 467813570d6526f286dc9b60de0b07d7930c631d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=B3=A2?= Date: Fri, 4 Nov 2022 10:10:02 +0800 Subject: [PATCH 04/24] =?UTF-8?q?Revert=20"fix=20bug=EF=BC=9A"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit d4fc0d7997d802ccfd449759ae56d09163e6396a. --- endpoint/src/msgque/strategy/hitfirst.rs | 32 +++++++++++------------- 1 file changed, 15 insertions(+), 17 deletions(-) diff --git a/endpoint/src/msgque/strategy/hitfirst.rs b/endpoint/src/msgque/strategy/hitfirst.rs index 567cac9ab..6eef14b3d 100644 --- a/endpoint/src/msgque/strategy/hitfirst.rs +++ b/endpoint/src/msgque/strategy/hitfirst.rs @@ -106,32 +106,30 @@ impl HitFirstReader { // 获取下一个读取node id let next = self.cursor_current.fetch_add(1, Ordering::Relaxed); let cursor_idx = next % self.cursors.len(); - let cursor = self.cursors.get(cursor_idx).unwrap(); - let mut node_idx = cursor.node_idx.load(Ordering::Relaxed); - let node_hits = cursor.hits.fetch_add(1, Ordering::Relaxed); + let chits = self.cursors.get(cursor_idx).unwrap(); + let mut idx = chits.node_idx.load(Ordering::Relaxed); + let hits = chits.hits.fetch_add(1, Ordering::Relaxed); // 如果当前位置hits数太大，也偏移一个位置 - if node_hits > MAX_HITS { - let next_idx = (node_idx + 1) % self.qnodes.len(); - if let Ok(_c) = cursor.node_idx.compare_exchange( - node_idx, - next_idx, - Ordering::Relaxed, - Ordering::Relaxed, - ) { - cursor.hits.store(0, Ordering::Relaxed); + if hits > MAX_HITS { + if let Ok(_c) = + chits + .node_idx + .compare_exchange(idx, idx + 1, Ordering::Relaxed, Ordering::Relaxed) + { + chits.hits.store(0, Ordering::Relaxed); } - // cursor已偏移到新位置，性能 - node_idx = next_idx; + // cursor已偏移到新位置 + idx += 1; } assert!( - node_idx < self.qnodes.len(), + idx < self.qnodes.len(), "idx:{}, qunodes:{:?}", - node_idx, + idx, self.qnodes ); - let node = self.qnodes.get(node_idx).unwrap(); + let node = self.qnodes.get(idx).unwrap(); log::debug!("+++ use common cursor:{}, {:?}", cursor_idx, self); node.id } From 4b8062e8212fe395624cdaeef5fae3134435f750 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=B3=A2?= Date: Fri, 4 Nov 2022 10:14:56 +0800 Subject: [PATCH 05/24] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dmq=E8=8A=82=E7=82=B9?= =?UTF-8?q?=E7=B4=A2=E5=BC=95=E8=8E=B7=E5=8F=96=E7=9A=84bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- endpoint/src/msgque/strategy/hitfirst.rs | 33 +++++++++++++----------- 1 file changed, 18 insertions(+), 15 deletions(-) diff --git a/endpoint/src/msgque/strategy/hitfirst.rs b/endpoint/src/msgque/strategy/hitfirst.rs index 6eef14b3d..6307c0d60 100644 --- a/endpoint/src/msgque/strategy/hitfirst.rs +++ b/endpoint/src/msgque/strategy/hitfirst.rs @@ -106,30 +106,33 @@ impl HitFirstReader { // 获取下一个读取node id let next = self.cursor_current.fetch_add(1, Ordering::Relaxed); let cursor_idx = next % self.cursors.len(); - let chits = self.cursors.get(cursor_idx).unwrap(); - let mut idx = chits.node_idx.load(Ordering::Relaxed); - let hits = chits.hits.fetch_add(1, Ordering::Relaxed); + let cursor = self.cursors.get(cursor_idx).unwrap(); + let mut node_idx = cursor.node_idx.load(Ordering::Relaxed); + let node_hits = cursor.hits.fetch_add(1, Ordering::Relaxed); // 如果当前位置hits数太大，也偏移一个位置 - if hits > MAX_HITS { - if let Ok(_c) = - chits - .node_idx - .compare_exchange(idx, idx + 1, Ordering::Relaxed, Ordering::Relaxed) - { - chits.hits.store(0, Ordering::Relaxed); + if node_hits > MAX_HITS { + let next_idx = (node_idx + 1) % self.qnodes.len(); + if let Ok(_c) = cursor.node_idx.compare_exchange( + node_idx, + next_idx, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + cursor.hits.store(0, Ordering::Relaxed); } - // cursor已偏移到新位置 - idx += 1; + + // cursor已偏移到新位置，性能 + node_idx = next_idx; } assert!( - idx < self.qnodes.len(), + node_idx < self.qnodes.len(), "idx:{}, qunodes:{:?}", - idx, + node_idx, self.qnodes ); - let node = self.qnodes.get(idx).unwrap(); + let node = self.qnodes.get(node_idx).unwrap(); log::debug!("+++ use common cursor:{}, {:?}", cursor_idx, self); node.id } From 252c47cc9bb7d5c76ffcc8c9ab0f83b79957af05 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 11:15:57 +0800 Subject: [PATCH 06/24] =?UTF-8?q?hot=20bug=20fix:=20=E5=9C=A8cacheservice?= =?UTF-8?q?=E4=B8=AD=EF=BC=8C=E4=BD=BF=E7=94=A8=E5=89=8Dn=E4=B8=AA?= =?UTF-8?q?=E5=85=83=E7=B4=A0=E8=BF=9B=E8=A1=8C=E8=AE=BF=E9=97=AE=E6=97=B6?= =?UTF-8?q?=EF=BC=8C=E7=AE=80=E5=8D=95=E7=9A=841+l1.len()=EF=BC=8C?= =?UTF-8?q?=E4=BD=86=E5=9C=A8=E8=AE=A1=E7=AE=97=E5=AE=9E=E9=99=85=E9=95=BF?= =?UTF-8?q?=E5=BA=A6=E6=97=B6=E8=BF=9B=E8=A1=8C=E4=BA=86=E5=8E=BB=E9=87=8D?= =?UTF-8?q?=E3=80=82=E5=9C=A8=E9=85=8D=E7=BD=AE=E4=B8=AD=E5=A6=82=E6=9E=9C?= =?UTF-8?q?=E6=9C=89=E9=87=8D=E5=A4=8D=E9=85=8D=E7=BD=AE=EF=BC=8C=E5=8F=AF?= =?UTF-8?q?=E8=83=BD=E5=AF=BC=E8=87=B4=E8=B6=8A=E7=95=8Cpanic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- endpoint/src/cacheservice/config.rs | 12 +++++++----- endpoint/src/cacheservice/topo.rs | 3 +-- sharding/src/select/by_distance.rs | 1 + 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/endpoint/src/cacheservice/config.rs b/endpoint/src/cacheservice/config.rs index 164cbcdfc..507017e0b 100644 --- a/endpoint/src/cacheservice/config.rs +++ b/endpoint/src/cacheservice/config.rs @@ -38,9 +38,9 @@ pub struct Namespace { } impl Namespace { - pub(crate) fn local_len(&self) -> usize { - 1 + self.master_l1.len() - } + //pub(crate) fn local_len(&self) -> usize { + // 1 + self.master_l1.len() + //} //pub(crate) fn is_static_hash(&self) -> bool { // match self.distribution.as_str() { // "modula" => true, @@ -83,17 +83,19 @@ impl Namespace { return true; } // 确保master在第0个位置 - pub(super) fn take_backends(self) -> Vec> { + // 返回master + master_l1的数量作为local + pub(super) fn take_backends(self) -> (usize, Vec>) { assert!(self.master.len() > 0); use ds::vec::Add; let mut backends = Vec::with_capacity(2 + self.master_l1.len() + self.slave_l1.len()); backends.add(self.master); self.master_l1.into_iter().for_each(|v| backends.add(v)); + let local = backends.len(); if self.slave.len() > 0 { backends.add(self.slave); } self.slave_l1.into_iter().for_each(|v| backends.add(v)); - backends + (local, backends) } //pub(super) fn timeout_master(&self) -> Duration { // Duration::from_millis(200.max(self.timeout_ms_master as u64)) diff --git a/endpoint/src/cacheservice/topo.rs b/endpoint/src/cacheservice/topo.rs index a914c12da..de8d6a1a2 100644 --- a/endpoint/src/cacheservice/topo.rs +++ b/endpoint/src/cacheservice/topo.rs @@ -201,9 +201,8 @@ where use discovery::distance::{Balance, ByDistance}; let master = ns.master.clone(); - let mut local_len = ns.local_len(); let local = ns.local_affinity; - let mut backends = ns.take_backends(); + let (mut local_len, mut backends) = ns.take_backends(); //let local = true; if local { backends.balance(&master); diff --git a/sharding/src/select/by_distance.rs b/sharding/src/select/by_distance.rs index 2523d99a2..2aad76afa 100644 --- a/sharding/src/select/by_distance.rs +++ b/sharding/src/select/by_distance.rs @@ -31,6 +31,7 @@ impl Distance { } // 只取前n个进行批量随机访问 pub fn topn(&mut self, n: usize) { + assert!(n > 0 && n <= self.len(), "n: {}, len:{}", n, self.len()); self.len_local = n as u16; let batch = 1024usize; // 最小是1，最大是65536 From 261cedca4f304a998a4666c0d420ef407a9bfe1d Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 11:32:55 +0800 Subject: [PATCH 07/24] =?UTF-8?q?mem=20buf=20resize:=20=E5=86=85=E5=AD=98?= =?UTF-8?q?=E5=9B=9E=E6=94=B6=E7=AD=96=E7=95=A5=E8=B0=83=E6=95=B4=E3=80=82?= =?UTF-8?q?1.=20=E4=BB=8E=E6=9C=89=E6=95=B0=E6=8D=AE=E5=86=99=E5=85=A5?= =?UTF-8?q?=E5=8D=B3=E8=A7=A6=E5=8F=91=E5=9B=9E=E6=94=B6=E8=B0=83=E6=95=B4?= =?UTF-8?q?=E4=B8=BA=E5=AE=9A=E6=9C=9F=E6=8C=89=E9=9C=80=E5=9B=9E=E6=94=B6?= =?UTF-8?q?=EF=BC=9B2.=20=E4=BB=8E=E5=8E=9F=E6=9C=89=E7=9A=84=E6=AF=8F?= =?UTF-8?q?=E6=AC=A1=E8=B0=83=E6=95=B4=E4=B8=BA=E5=8E=9F=E6=9D=A5=E7=9A=84?= =?UTF-8?q?=E4=B8=80=E5=8D=8A=EF=BC=8C=E8=B0=83=E6=95=B4=E4=B8=BA=E6=9C=80?= =?UTF-8?q?=E8=BF=9110=E5=88=86=E9=92=9F=E6=9C=80=E5=A4=A7=E5=80=BC?= =?UTF-8?q?=E7=9A=842=E5=80=8D=EF=BC=9B3.=20=E9=99=8D=E4=BD=8E=E4=B8=8D?= =?UTF-8?q?=E5=BF=85=E8=A6=81=E7=9A=84poll=5Ftick=E6=B6=88=E8=80=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent/src/http.rs | 4 +- agent/src/service.rs | 6 +-- ds/src/mem/guarded.rs | 2 +- ds/src/mem/policy.rs | 83 ++++++++++++++---------------- ds/src/mem/resized.rs | 26 +++++++--- protocol/src/write.rs | 10 ++++ rt/src/entry.rs | 46 +++++++---------- rt/src/stream/mod.rs | 27 +++++++--- sharding/src/select/by_distance.rs | 2 +- stream/src/buffer.rs | 23 +++++---- stream/src/handler.rs | 20 +++++-- stream/src/lib.rs | 5 +- stream/src/pipeline.rs | 11 ++-- 13 files changed, 150 insertions(+), 115 deletions(-) diff --git a/agent/src/http.rs b/agent/src/http.rs index 7ff7c9e70..6ae71ef71 100644 --- a/agent/src/http.rs +++ b/agent/src/http.rs @@ -15,9 +15,7 @@ pub(super) fn start_http_server(ctx: &context::Context) { c.log_level = LogLevel::Critical; c.workers = 4; let mut rocket = rocket::custom(c); - { - rocket = crate::console::init_routes(rocket, ctx); - } + rocket = crate::console::init_routes(rocket, ctx); rocket = crate::prometheus::init_routes(rocket); //rocket = rocket.attach(rocket_async_compression::Compression::fairing()); rt::spawn(async { diff --git a/agent/src/service.rs b/agent/src/service.rs index 4f15d9582..62a74314d 100644 --- a/agent/src/service.rs +++ b/agent/src/service.rs @@ -9,11 +9,9 @@ use ds::chan::Sender; use metrics::Path; use protocol::{Parser, Result}; use stream::pipeline::copy_bidirectional; -use stream::Builder; -use stream::StreamMetrics; +use stream::{Backend, Builder, Request, StreamMetrics}; -use stream::Request; -type Endpoint = Arc>; +type Endpoint = Arc>; type Topology = endpoint::Topology, Endpoint, Request, Parser>; // 一直侦听，直到成功侦听或者取消侦听（当前尚未支持取消侦听） // 1. 尝试侦听之前，先确保服务配置信息已经更新完成 diff --git a/ds/src/mem/guarded.rs b/ds/src/mem/guarded.rs index f669978c6..eaaa63514 100644 --- a/ds/src/mem/guarded.rs +++ b/ds/src/mem/guarded.rs @@ -122,7 +122,7 @@ impl MemGuard { unsafe { assert_eq!((&*guard).load(Ordering::Acquire), 0) }; Self { mem: data, - guard: guard, + guard, cap: 0, } } diff --git a/ds/src/mem/policy.rs b/ds/src/mem/policy.rs index b1367894d..819e70c6e 100644 --- a/ds/src/mem/policy.rs +++ b/ds/src/mem/policy.rs @@ -1,11 +1,10 @@ const BUF_MIN: usize = 1024; -use std::time::{Duration, Instant}; +use std::time::Instant; // 内存需要缩容时的策略 // 为了避免频繁的缩容，需要设置一个最小频繁，通常使用最小间隔时间 pub struct MemPolicy { - ticks: usize, last: Instant, // 上一次tick返回true的时间 - secs: u16, // 每两次tick返回true的最小间隔时间 + max: usize, // 最近一个周期内，最大的内存使用量。 // 下面两个变量为了输出日志 trace: trace::Trace, @@ -19,13 +18,11 @@ impl MemPolicy { Self::with_direction("rx") } pub fn with_direction(direction: &'static str) -> Self { - Self::from(Duration::from_secs(600), direction) + Self::from(direction) } - fn from(delay: Duration, direction: &'static str) -> Self { - let secs = delay.as_secs().max(1).min(u16::MAX as u64) as u16; + fn from(direction: &'static str) -> Self { Self { - ticks: 0, - secs, + max: 0, last: Instant::now(), trace: direction.into(), } @@ -34,35 +31,35 @@ impl MemPolicy { pub fn need_grow(&self, len: usize, cap: usize, reserve: usize) -> bool { len + reserve > cap } - // 每隔31次进行一次check - // 连续self.secs秒check返回true，则需要缩容 #[inline] - pub fn need_shrink(&mut self, len: usize, cap: usize) -> bool { - // 长度 * 4 >= cap，说明利用率大于25% - if len >= (cap >> 2) || cap <= BUF_MIN { - self.reset(); - return false; - } - self.ticks += 1; - // 定期检查。 - const TICKS: usize = 31; - if self.ticks & TICKS != 0 { - return false; - } - if self.ticks == TICKS + 1 { - self.last = Instant::now(); - return false; + pub fn check_shrink(&mut self, len: usize, _cap: usize) { + if self.max < len { + self.reset(len); } - if self.last.elapsed().as_secs() <= self.secs as u64 { - return false; - } - true } - #[inline(always)] - fn reset(&mut self) { - if self.ticks > 0 { - self.ticks = 0; + fn reset(&mut self, len: usize) { + self.max = len; + self.last = Instant::now(); + } + // 在一个周期内，max * 4 <= cap. 则需要缩容 + #[inline] + pub fn need_shrink(&mut self, len: usize, cap: usize) -> bool { + log::debug!("need_shrink: len: {}, cap: {} => {}", len, cap, self); + if cap > BUF_MIN { + self.check_shrink(len, cap); + // 600秒对于大部分在线业务，足够得出稳定的max值。 + if self.last.elapsed().as_secs() >= 600 { + if self.max < (cap >> 2) { + // 只有当前buff size是0时才触发缩容。 + return true; + } + // 重新开始一个周期 + self.reset(len); + } + } else { + self.reset(len); } + false } // 确认缩容的size // 1. 最小值为 len + reserve的1.25倍 @@ -70,25 +67,22 @@ impl MemPolicy { // 3. 至少为BUF_MIN // 4. 2的指数倍 #[inline] - pub fn grow(&self, len: usize, cap: usize, reserve: usize) -> usize { + pub fn grow(&mut self, len: usize, cap: usize, reserve: usize) -> usize { let new = ((5 * (len + reserve)) / 4) .max(cap) .max(BUF_MIN) .next_power_of_two(); log::info!("grow: {} {} > {} => {} {}", len, reserve, cap, new, self); + self.max = 0; new } - // 确认缩容的size: - // 1. 当前容量的一半 - // 2. 最小值为MIN_BUF - // 3. 最小值为len - // 4. 取2的指数倍 - // 注意：返回值可能比输入的cap大. 但在判断need_shrink时，会判断len * 4 < cap, 所以不会出现len * 4 < cap, 但是cap / 2 < len的情况 #[inline] pub fn shrink(&mut self, len: usize, cap: usize) -> usize { - let new = (cap / 2).max(BUF_MIN).max(len).next_power_of_two(); + assert!(self.max < cap, "{}", self); + let new = (self.max * 2).max(BUF_MIN).max(len).next_power_of_two(); log::info!("shrink: {} < {} => {} {}", len, cap, new, self); - self.ticks = 0; + assert!(new >= len); + self.max = 0; new } } @@ -103,10 +97,9 @@ impl Display for MemPolicy { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { write!( f, - "buf policy: ticks: {} last: {:?} secs: {}{:?}", - self.ticks, + "buf policy: max: {} last: {:?} {:?}", + self.max, self.last.elapsed(), - self.secs, self.trace ) } diff --git a/ds/src/mem/resized.rs b/ds/src/mem/resized.rs index 07dd38e0f..2336f9b43 100644 --- a/ds/src/mem/resized.rs +++ b/ds/src/mem/resized.rs @@ -62,17 +62,16 @@ impl ResizedRingBuffer { // 有数写入时，判断是否需要缩容 #[inline] pub fn advance_write(&mut self, n: usize) { - self.inner.advance_write(n); - // 判断是否需要缩容 - if self.policy.need_shrink(self.len(), self.cap()) { - let new = self.policy.shrink(self.len(), self.cap()); - self.resize(new); + if n > 0 { + self.inner.advance_write(n); + self.policy.check_shrink(self.len(), self.cap()); } + // 判断是否需要缩容 } #[inline] fn resize(&mut self, cap: usize) { - assert!(cap <= self.max as usize); - assert!(cap >= self.min as usize); + assert!(cap <= self.max as usize, "{} > {}", cap, self.max); + assert!(cap >= self.min as usize, "{} < {}", cap, self.min); let new = self.inner.resize(cap); let old = std::mem::replace(&mut self.inner, new); self.max_processed = old.writtened(); @@ -110,12 +109,23 @@ impl ResizedRingBuffer { self.resize(new); } } + #[inline] + pub fn shrink(&mut self) { + if self.policy.need_shrink(self.len(), self.cap()) { + let new = self.policy.shrink(self.len(), self.cap()); + self.resize(new); + } + } } use std::fmt::{self, Display, Formatter}; impl Display for ResizedRingBuffer { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - write!(f, "rrb:(inner:{}, old:{:?})", self.inner, self.old) + write!( + f, + "rrb:(inner:{}, old:{:?}) policy:{}", + self.inner, self.old, self.policy + ) } } diff --git a/protocol/src/write.rs b/protocol/src/write.rs index 7588c4210..e7bd0af4e 100644 --- a/protocol/src/write.rs +++ b/protocol/src/write.rs @@ -1,5 +1,6 @@ use super::Result; pub trait Writer { + fn cap(&self) -> usize; fn pending(&self) -> usize; // 写数据，一次写完 fn write(&mut self, data: &[u8]) -> Result<()>; @@ -38,8 +39,13 @@ pub trait Writer { } Ok(()) } + fn shrink(&mut self); } impl Writer for Vec { + #[inline] + fn cap(&self) -> usize { + self.capacity() + } #[inline] fn pending(&self) -> usize { self.len() @@ -54,4 +60,8 @@ impl Writer for Vec { self.push(v); Ok(()) } + #[inline] + fn shrink(&mut self) { + todo!("should not call shrink"); + } } diff --git a/rt/src/entry.rs b/rt/src/entry.rs index 77e03d7ee..dc66eb361 100644 --- a/rt/src/entry.rs +++ b/rt/src/entry.rs @@ -27,14 +27,10 @@ pub trait ReEnter { // true: 成功关闭，释放相关资源 // false: 还有资源未释放 fn close(&mut self) -> bool; - //#[inline] - //fn need_refresh(&self) -> bool { - // false - //} - #[inline] - fn refresh(&mut self) -> bool { - false - } + // 定期会调用，通常用来清理内存，更新数据等信息。 + // 返回true: 表示期待进行下一次调用 + // 返回false: 表示资源已经释放，不再需要调用。 + fn refresh(&mut self) -> bool; } pub trait Cancel { fn cancel(&mut self); @@ -65,7 +61,8 @@ pub struct Entry { ready: bool, refresh_tick: Interval, out: Option>, - //runs: usize, + refresh_next: bool, + last_refresh: Instant, } impl> + Unpin + ReEnter + Debug> Entry { #[inline] @@ -73,7 +70,7 @@ impl> + Unpin + ReEnter + Debug> Entry { let mut tick = interval(timeout.max(Duration::from_millis(50))); tick.set_missed_tick_behavior(MissedTickBehavior::Delay); - let mut refresh_tick = interval(Duration::from_secs(30)); + let mut refresh_tick = interval(Duration::from_secs(9)); refresh_tick.set_missed_tick_behavior(MissedTickBehavior::Delay); let m_reenter = Path::base().rtt("reenter10ms"); @@ -87,18 +84,17 @@ impl> + Unpin + ReEnter + Debug> Entry { ready: false, out: None, refresh_tick, - //runs: 0, + refresh_next: false, + last_refresh: Instant::now(), } } #[inline] fn poll_run(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let now = Instant::now(); - //if self.inner.need_refresh() { - // self.runs += 1; - // if self.runs & 15 == 0 { - // self.inner.refresh(); - // } - //} + if (now - self.last_refresh).as_secs() > 3 { + self.refresh_next = self.inner.refresh(); + self.last_refresh = now; + } let (tx, rx) = (self.inner.num_tx(), self.inner.num_rx()); if tx > rx { @@ -113,7 +109,6 @@ impl> + Unpin + ReEnter + Debug> Entry { self.last_rx = now; } - // 发现异常立即返回处理 let ret = Pin::new(&mut self.inner).poll(cx)?; let (tx_post, rx_post) = (self.inner.num_tx(), self.inner.num_rx()); if tx_post > rx_post { @@ -122,20 +117,17 @@ impl> + Unpin + ReEnter + Debug> Entry { if rx_post > rx { self.last_rx = self.last; } - loop { - ready!(self.tick.poll_tick(cx)); - } + ready!(self.tick.poll_tick(cx)); + self.tick.reset(); } else { - //if self.inner.need_refresh() { - // 如果需要刷新，则不阻塞在原有的pending上 - if self.inner.refresh() && ret.is_pending() { - loop { + if ret.is_pending() { + if self.refresh_next { ready!(self.refresh_tick.poll_tick(cx)); + self.refresh_tick.reset(); } } - //} - ret.map(|rs| Ok(rs)) } + ret.map(|r| Ok(r)) } } diff --git a/rt/src/stream/mod.rs b/rt/src/stream/mod.rs index e69b9e6e8..332d95dde 100644 --- a/rt/src/stream/mod.rs +++ b/rt/src/stream/mod.rs @@ -101,6 +101,10 @@ impl AsyncWrite for Stream { } impl protocol::Writer for Stream { + #[inline] + fn cap(&self) -> usize { + self.buf.capacity() + } #[inline] fn pending(&self) -> usize { self.buf.len() @@ -124,6 +128,19 @@ impl protocol::Writer for Stream { self.write_to_buf = hint; } } + #[inline] + fn shrink(&mut self) { + if self + .policy + .need_shrink(self.buf.len(), self.buf.capacity()) + { + let old = self.buf.capacity(); + let new = self.policy.shrink(self.buf.len(), old); + self.buf.shrink_to(new); + assert_eq!(new, self.buf.capacity(), "{}/{}", new, self.buf.capacity()); + self.buf_tx -= (old - new) as isize; + } + } } impl Stream { #[inline(always)] @@ -143,13 +160,9 @@ impl Stream { // 所有数据flush完了之后，尝试进行缩容 #[inline] fn clear(&mut self) { - if self.policy.need_shrink(self.buf.len(), self.buf.capacity()) { - let old = self.buf.capacity(); - let new = self.policy.shrink(self.buf.len(), old); - self.buf.shrink_to(new); - assert_eq!(new, self.buf.capacity(), "{}/{}", new, self.buf.capacity()); - self.buf_tx -= (old - new) as isize; - } + self.policy + .check_shrink(self.buf.len(), self.buf.capacity()); + self.idx = 0; unsafe { self.buf.set_len(0) }; } diff --git a/sharding/src/select/by_distance.rs b/sharding/src/select/by_distance.rs index 2aad76afa..acdf99beb 100644 --- a/sharding/src/select/by_distance.rs +++ b/sharding/src/select/by_distance.rs @@ -68,7 +68,7 @@ impl Distance { } else { (self.seq.fetch_add(1, Relaxed) >> self.batch_shift as usize) % self.local_len() }; - assert!(idx < self.replicas.len()); + assert!(idx < self.len(), "{} >= {}", idx, self.len()); idx } // 只从local获取 diff --git a/stream/src/buffer.rs b/stream/src/buffer.rs index 51fafd065..00e96f252 100644 --- a/stream/src/buffer.rs +++ b/stream/src/buffer.rs @@ -48,7 +48,6 @@ where let out = Pin::new(&mut **client).poll_read(cx, &mut rb); let r = rb.capacity() - rb.remaining(); if r > 0 { - // log::debug!("{} bytes received ==> {:?}", r, &buf[0..r]); log::debug!("{} bytes received", r); } *n += r; @@ -106,11 +105,11 @@ impl From for StreamGuard { impl StreamGuard { #[inline] pub fn init(init: usize) -> Self { - const MIN: usize = 1024; // buffer最大从4M调整到64M，观察CPU、Mem fishermen 2022.5.23 - const MAX: usize = 64 << 20; - let init = init.max(MIN).min(MAX); - Self::with(MIN, MAX, init) + let min = crate::MIN_BUFFER_SIZE; + let max = crate::MAX_BUFFER_SIZE; + let init = init.max(min).min(max); + Self::with(min, max, init) } #[inline] pub fn new() -> Self { @@ -145,17 +144,21 @@ impl StreamGuard { { self.buf.write(r) } - //#[inline] - //pub fn cap(&self) -> usize { - // self.buf.cap() - //} + #[inline] + pub fn shrink(&mut self) { + self.buf.shrink(); + } + #[inline] + pub fn cap(&self) -> usize { + self.buf.cap() + } } use std::fmt::{self, Debug, Display, Formatter}; impl Display for StreamGuard { #[inline] fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - write!(f, "ctx:{} StreamGuard :{}", self.ctx, self.buf,) + write!(f, "ctx:{} {}", self.ctx, self.buf) } } impl Debug for StreamGuard { diff --git a/stream/src/handler.rs b/stream/src/handler.rs index 7ae5b130e..a3009781c 100644 --- a/stream/src/handler.rs +++ b/stream/src/handler.rs @@ -4,7 +4,7 @@ use std::pin::Pin; use std::task::{Context, Poll}; use ds::chan::mpsc::Receiver; -use protocol::{Error, Protocol, Request, Result, Stream}; +use protocol::{Error, Protocol, Request, Result, Stream, Writer}; use std::task::ready; use tokio::io::{AsyncRead, AsyncWrite}; @@ -12,7 +12,7 @@ use crate::buffer::StreamGuard; use metrics::Metric; -pub(crate) struct Handler<'r, Req, P, S> { +pub struct Handler<'r, Req, P, S> { data: &'r mut Receiver, pending: VecDeque, @@ -29,7 +29,7 @@ pub(crate) struct Handler<'r, Req, P, S> { impl<'r, Req, P, S> Future for Handler<'r, Req, P, S> where Req: Request + Unpin, - S: AsyncRead + AsyncWrite + protocol::Writer + Unpin, + S: AsyncRead + AsyncWrite + Writer + Unpin, P: Protocol + Unpin, { type Output = Result<()>; @@ -71,6 +71,7 @@ where fn poll_request(&mut self, cx: &mut Context) -> Poll> { self.s.cache(self.data.size_hint() > 1); while let Some(req) = ready!(self.data.poll_recv(cx)) { + log::info!("request:{:?}", req); self.num_tx += 1; self.s.write_slice(req.data(), 0)?; match req.on_sent() { @@ -97,6 +98,7 @@ where self.num_rx += 1; // 统计请求耗时。 self.rtt += req.start_at().elapsed(); + log::info!("response: {:?}", cmd); req.on_complete(cmd); } } @@ -124,7 +126,7 @@ where } unsafe impl<'r, Req, P, S> Send for Handler<'r, Req, P, S> {} unsafe impl<'r, Req, P, S> Sync for Handler<'r, Req, P, S> {} -impl<'r, Req: Request, P, S: AsyncRead + AsyncWrite + Unpin> rt::ReEnter +impl<'r, Req: Request, P, S: AsyncRead + AsyncWrite + Unpin + Writer> rt::ReEnter for Handler<'r, Req, P, S> { #[inline] @@ -154,6 +156,14 @@ impl<'r, Req: Request, P, S: AsyncRead + AsyncWrite + Unpin> rt::ReEnter self.buf.try_gc() } + #[inline] + fn refresh(&mut self) -> bool { + log::debug!("handler:{:?}", self); + self.buf.try_gc(); + self.buf.shrink(); + self.s.shrink(); + self.buf.cap() + self.s.cap() > 4096 + } } use std::fmt::{self, Debug, Formatter}; @@ -162,7 +172,7 @@ impl<'r, Req, P, S> Debug for Handler<'r, Req, P, S> { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { write!( f, - "handler tx_seq:{} rx_seq:{} pending:{} {} buf:{:?}", + "handler tx_seq:{} rx_seq:{} p_req:{} {} buf:{:?}", self.num_tx, self.num_rx, self.pending.len(), diff --git a/stream/src/lib.rs b/stream/src/lib.rs index 92e11098a..e8e239057 100644 --- a/stream/src/lib.rs +++ b/stream/src/lib.rs @@ -1,5 +1,5 @@ pub mod buffer; -pub(crate) mod handler; +pub mod handler; pub mod pipeline; mod shards; pub use protocol::callback::*; @@ -23,3 +23,6 @@ pub(crate) mod checker; mod metric; pub use metric::CbMetrics as StreamMetrics; + +pub(crate) const MIN_BUFFER_SIZE: usize = 1024; +pub(crate) const MAX_BUFFER_SIZE: usize = 64 << 20; diff --git a/stream/src/pipeline.rs b/stream/src/pipeline.rs index 42e314d7b..64e4e18dc 100644 --- a/stream/src/pipeline.rs +++ b/stream/src/pipeline.rs @@ -275,8 +275,10 @@ impl Drop for CopyBidirectional { } use std::fmt::{self, Debug, Formatter}; -impl> rt::ReEnter - for CopyBidirectional +impl rt::ReEnter for CopyBidirectional +where + C: AsyncRead + AsyncWrite + Writer + Unpin, + T: TopologyCheck + Topology, { #[inline] fn close(&mut self) -> bool { @@ -329,7 +331,10 @@ impl 4096 } } impl Debug for CopyBidirectional { From a2f1a6c617b63d03c624c7e085dceefb2b9a2058 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 11:33:23 +0800 Subject: [PATCH 08/24] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E9=83=A8=E5=88=86struc?= =?UTF-8?q?t=E7=9A=84layout=E6=B5=8B=E8=AF=95=E7=94=A8=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/Cargo.toml | 2 ++ tests/src/layout.rs | 36 ++++++++++++++++++++++++++++++++++-- 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 7903f1fb3..b2c60a1a4 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -16,6 +16,8 @@ byteorder = "1.4.3" bmemcached = "0.5.0" assert-panic = "1.0.1" metrics = { path = "../metrics" } +endpoint = { path = "../endpoint" } +rt = { path = "../rt" } tokio = { version = "1.21.2", features = ["fs"] } redis = { version = "0.22.0", default-features = false, features = [] } diff --git a/tests/src/layout.rs b/tests/src/layout.rs index 1e0c406da..3998b7d2b 100644 --- a/tests/src/layout.rs +++ b/tests/src/layout.rs @@ -1,5 +1,25 @@ // 不要轻易变更这里面的测试用例，除非你知道你在做什么。拉相关同学进行方案评审。 -use std::{mem::size_of, sync::atomic::AtomicU32}; +use std::{ + mem::size_of, + sync::{atomic::AtomicU32, Arc}, +}; + +use protocol::Parser; +use stream::{Backend, Request}; +type Endpoint = Arc>; +type Topology = endpoint::Topology; + +type Stream = rt::Stream; + +type Handler<'r> = stream::handler::Handler<'r, Request, Parser, Stream>; + +type Builder = stream::Builder; +type CacheService = endpoint::cacheservice::topo::CacheService; +type RedisService = endpoint::redisservice::topo::RedisService; +type PhantomService = + endpoint::phantomservice::topo::PhantomService; +type MsgQue = endpoint::msgque::topo::MsgQue; + #[test] fn check_layout() { assert_eq!(32, size_of::()); @@ -11,9 +31,21 @@ fn check_layout() { ); assert_eq!(24, size_of::()); assert_eq!(1, size_of::()); - let guard_size = if cfg!(debug_assertions) { 240 } else { 200 }; + let guard_size = if cfg!(debug_assertions) { 232 } else { 192 }; assert_eq!(guard_size, size_of::()); assert_eq!(56, size_of::>()); assert_eq!(16, size_of::()); assert_eq!(64, size_of::()); + assert_eq!(1, size_of::()); + assert_eq!(48, size_of::>()); + let stream_size = if cfg!(debug_assertions) { 248 } else { 200 }; + assert_eq!(stream_size, size_of::()); + assert_eq!(560, size_of::>()); + assert_eq!(0, size_of::()); + + assert_eq!(400, size_of::()); + assert_eq!(96, size_of::()); + assert_eq!(264, size_of::()); + assert_eq!(192, size_of::()); + assert_eq!(392, size_of::()); } From ead42be3ec8d7f6acd7a3895fc1bd5b1168815cf Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 16:02:12 +0800 Subject: [PATCH 09/24] =?UTF-8?q?=E5=88=A0=E9=99=A4=E9=83=A8=E5=88=86?= =?UTF-8?q?=E4=B8=8D=E5=BF=85=E8=A6=81=E7=9A=84=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ds/src/mem/ring_slice.rs | 90 ++++++++++++++++++++-------------------- 1 file changed, 45 insertions(+), 45 deletions(-) diff --git a/ds/src/mem/ring_slice.rs b/ds/src/mem/ring_slice.rs index cba918ff5..fbd561f31 100644 --- a/ds/src/mem/ring_slice.rs +++ b/ds/src/mem/ring_slice.rs @@ -91,26 +91,26 @@ impl RingSlice { self.ptr().offset(oft as isize) } - pub fn split(&self, splitter: &[u8]) -> Vec { - let mut pos = 0 as usize; - let mut result: Vec = vec![]; - loop { - let new_pos = self.find_sub(pos, splitter); - if new_pos.is_none() { - if pos < self.len() { - result.push(self.sub_slice(pos, self.len() - pos)); - } - return result; - } else { - let new_pos = new_pos.unwrap(); - result.push(self.sub_slice(pos, new_pos)); - if new_pos + splitter.len() == self.end - self.start { - return result; - } - pos = pos + new_pos + splitter.len(); - } - } - } + //pub fn split(&self, splitter: &[u8]) -> Vec { + // let mut pos = 0 as usize; + // let mut result: Vec = vec![]; + // loop { + // let new_pos = self.find_sub(pos, splitter); + // if new_pos.is_none() { + // if pos < self.len() { + // result.push(self.sub_slice(pos, self.len() - pos)); + // } + // return result; + // } else { + // let new_pos = new_pos.unwrap(); + // result.push(self.sub_slice(pos, new_pos)); + // if new_pos + splitter.len() == self.end - self.start { + // return result; + // } + // pos = pos + new_pos + splitter.len(); + // } + // } + //} #[inline] pub fn find(&self, offset: usize, b: u8) -> Option { for i in offset..self.len() { @@ -120,31 +120,31 @@ impl RingSlice { } None } - // 从offset开始，查找s是否存在 - // 最坏时间复杂度 O(self.len() * s.len()) - // 但通常在协议处理过程中，被查的s都是特殊字符，而且s的长度通常比较小，因为时间复杂度会接近于O(self.len()) - pub fn find_sub(&self, offset: usize, s: &[u8]) -> Option { - if self.start + offset + s.len() > self.end { - return None; - } - let mut i = 0 as usize; - let i_cap = self.len() - s.len() - offset; - while i <= i_cap { - let mut found_len = 0 as usize; - for j in 0..s.len() { - if self.read_u8(offset + i + j) != s[j] { - i += 1; - break; - } else { - found_len = found_len + 1; - } - } - if found_len == s.len() { - return Some(i); - } - } - None - } + //// 从offset开始，查找s是否存在 + //// 最坏时间复杂度 O(self.len() * s.len()) + //// 但通常在协议处理过程中，被查的s都是特殊字符，而且s的长度通常比较小，因为时间复杂度会接近于O(self.len()) + //pub fn find_sub(&self, offset: usize, s: &[u8]) -> Option { + // if self.start + offset + s.len() > self.end { + // return None; + // } + // let mut i = 0 as usize; + // let i_cap = self.len() - s.len() - offset; + // while i <= i_cap { + // let mut found_len = 0 as usize; + // for j in 0..s.len() { + // if self.read_u8(offset + i + j) != s[j] { + // i += 1; + // break; + // } else { + // found_len = found_len + 1; + // } + // } + // if found_len == s.len() { + // return Some(i); + // } + // } + // None + //} // 查找是否存在 '\r\n' ，返回匹配的第一个字节地址 #[inline] pub fn find_lf_cr(&self, offset: usize) -> Option { From ccbe21482021e62d12635c3095d5fe8e0b529899 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 16:04:04 +0800 Subject: [PATCH 10/24] =?UTF-8?q?mem=20resize:=20=E5=A6=82=E6=9E=9C?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=9C=AA=E8=AF=BB=E5=8F=96=E5=88=B0=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E5=AE=8C=E6=95=B4=E7=9A=84=E5=8C=85=EF=BC=8C=E5=88=99?= =?UTF-8?q?=E5=B0=9D=E8=AF=95=E8=BF=9B=E8=A1=8C=E4=B8=80=E6=AC=A1grow?= =?UTF-8?q?=EF=BC=8C=E4=BC=98=E5=8C=96=E5=A4=A7value=E7=9A=84=E5=86=85?= =?UTF-8?q?=E5=AD=98=E7=A2=8E=E7=89=87=E4=B8=8E=E6=80=A7=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ds/src/mem/policy.rs | 5 +++-- ds/src/mem/resized.rs | 14 ++++---------- protocol/src/memcache/binary/mod.rs | 3 +++ protocol/src/parser.rs | 1 + stream/src/buffer.rs | 18 +++++++----------- stream/src/lib.rs | 3 ++- tests/src/layout.rs | 5 +++-- 7 files changed, 23 insertions(+), 26 deletions(-) diff --git a/ds/src/mem/policy.rs b/ds/src/mem/policy.rs index 819e70c6e..8ab5acf8f 100644 --- a/ds/src/mem/policy.rs +++ b/ds/src/mem/policy.rs @@ -1,4 +1,4 @@ -const BUF_MIN: usize = 1024; +const BUF_MIN: usize = 2 * 1024; use std::time::Instant; // 内存需要缩容时的策略 // 为了避免频繁的缩容，需要设置一个最小频繁，通常使用最小间隔时间 @@ -14,7 +14,7 @@ impl MemPolicy { pub fn tx() -> Self { Self::with_direction("tx") } - pub fn rx() -> Self { + pub fn rx(_min: usize, _max: usize) -> Self { Self::with_direction("rx") } pub fn with_direction(direction: &'static str) -> Self { @@ -29,6 +29,7 @@ impl MemPolicy { } #[inline(always)] pub fn need_grow(&self, len: usize, cap: usize, reserve: usize) -> bool { + log::debug!("need_grow: len={}, cap={}, reserve={}", len, cap, reserve); len + reserve > cap } #[inline] diff --git a/ds/src/mem/resized.rs b/ds/src/mem/resized.rs index 2336f9b43..9b50c4a8f 100644 --- a/ds/src/mem/resized.rs +++ b/ds/src/mem/resized.rs @@ -11,8 +11,6 @@ pub struct ResizedRingBuffer { old: Vec, inner: RingBuffer, // 下面的用来做扩缩容判断 - min: u32, - max: u32, on_change: Callback, policy: MemPolicy, } @@ -41,16 +39,14 @@ impl ResizedRingBuffer { init: usize, mut cb: F, ) -> Self { - assert!(min <= init && init <= max); + assert!(min <= max && init <= max); cb(0, init as isize); Self { max_processed: std::usize::MAX, old: Vec::new(), inner: RingBuffer::with_capacity(init), - min: min as u32, - max: max as u32, on_change: Box::new(cb), - policy: MemPolicy::rx(), + policy: MemPolicy::rx(min, max), } } // 需要写入数据时，判断是否需要扩容 @@ -70,8 +66,6 @@ impl ResizedRingBuffer { } #[inline] fn resize(&mut self, cap: usize) { - assert!(cap <= self.max as usize, "{} > {}", cap, self.max); - assert!(cap >= self.min as usize, "{} < {}", cap, self.min); let new = self.inner.resize(cap); let old = std::mem::replace(&mut self.inner, new); self.max_processed = old.writtened(); @@ -102,8 +96,8 @@ impl ResizedRingBuffer { self.grow(data.len()); self.inner.write(data) } - #[inline(always)] - fn grow(&mut self, reserve: usize) { + #[inline] + pub fn grow(&mut self, reserve: usize) { if self.policy.need_grow(self.len(), self.cap(), reserve) { let new = self.policy.grow(self.len(), self.cap(), reserve); self.resize(new); diff --git a/protocol/src/memcache/binary/mod.rs b/protocol/src/memcache/binary/mod.rs index 1311d8594..b224c1fe1 100644 --- a/protocol/src/memcache/binary/mod.rs +++ b/protocol/src/memcache/binary/mod.rs @@ -29,6 +29,7 @@ impl Protocol for MemcacheBinary { req.check_request()?; let packet_len = req.packet_len(); if req.len() < packet_len { + data.reserve(packet_len - req.len()); break; } @@ -67,6 +68,8 @@ impl Protocol for MemcacheBinary { assert!(!r.status_ok(), "rsp: {:?}", r); data.ignore(pl); } + } else { + data.reserve(pl - len); } } Ok(None) diff --git a/protocol/src/parser.rs b/protocol/src/parser.rs index 4031d5447..58293dad9 100644 --- a/protocol/src/parser.rs +++ b/protocol/src/parser.rs @@ -92,6 +92,7 @@ pub trait Stream { fn context(&mut self) -> &mut u64; // 用于保存下一个cmd需要使用的hash fn reserved_hash(&mut self) -> &mut i64; + fn reserve(&mut self, r: usize); } pub trait Builder { diff --git a/stream/src/buffer.rs b/stream/src/buffer.rs index 00e96f252..7c2e7eb2c 100644 --- a/stream/src/buffer.rs +++ b/stream/src/buffer.rs @@ -63,14 +63,6 @@ pub struct StreamGuard { buf: GuardedBuffer, } impl protocol::Stream for StreamGuard { - //#[inline] - //fn update(&mut self, idx: usize, val: u8) { - // self.buf.update(idx, val); - //} - //#[inline] - //fn at(&self, idx: usize) -> u8 { - // self.buf.at(idx) - //} #[inline] fn take(&mut self, n: usize) -> MemGuard { self.buf.take(n) @@ -91,6 +83,10 @@ impl protocol::Stream for StreamGuard { fn reserved_hash(&mut self) -> &mut i64 { &mut self.reserved_hash } + #[inline] + fn reserve(&mut self, r: usize) { + self.buf.grow(r); + } } impl From for StreamGuard { #[inline] @@ -108,12 +104,12 @@ impl StreamGuard { // buffer最大从4M调整到64M，观察CPU、Mem fishermen 2022.5.23 let min = crate::MIN_BUFFER_SIZE; let max = crate::MAX_BUFFER_SIZE; - let init = init.max(min).min(max); - Self::with(min, max, init) + Self::with(min, max, init.min(max)) } #[inline] pub fn new() -> Self { - Self::init(1024) + // 初始化为0，延迟分配内存 + Self::init(0) } #[inline] fn with(min: usize, max: usize, init: usize) -> Self { diff --git a/stream/src/lib.rs b/stream/src/lib.rs index e8e239057..4810ba5c7 100644 --- a/stream/src/lib.rs +++ b/stream/src/lib.rs @@ -24,5 +24,6 @@ pub(crate) mod checker; mod metric; pub use metric::CbMetrics as StreamMetrics; -pub(crate) const MIN_BUFFER_SIZE: usize = 1024; +// 最小2K，至少容纳一个MTU +pub(crate) const MIN_BUFFER_SIZE: usize = 1024 * 2; pub(crate) const MAX_BUFFER_SIZE: usize = 64 << 20; diff --git a/tests/src/layout.rs b/tests/src/layout.rs index 3998b7d2b..167eb239d 100644 --- a/tests/src/layout.rs +++ b/tests/src/layout.rs @@ -31,7 +31,7 @@ fn check_layout() { ); assert_eq!(24, size_of::()); assert_eq!(1, size_of::()); - let guard_size = if cfg!(debug_assertions) { 232 } else { 192 }; + let guard_size = if cfg!(debug_assertions) { 224 } else { 184 }; assert_eq!(guard_size, size_of::()); assert_eq!(56, size_of::>()); assert_eq!(16, size_of::()); @@ -40,7 +40,8 @@ fn check_layout() { assert_eq!(48, size_of::>()); let stream_size = if cfg!(debug_assertions) { 248 } else { 200 }; assert_eq!(stream_size, size_of::()); - assert_eq!(560, size_of::>()); + let handler_size = if cfg!(debug_assertions) { 552 } else { 464 }; + assert_eq!(handler_size, size_of::>()); assert_eq!(0, size_of::()); assert_eq!(400, size_of::()); From d4c376fd385a2a2e2b13d37eeef031ae2af0a527 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 16:27:00 +0800 Subject: [PATCH 11/24] =?UTF-8?q?=E5=88=A0=E9=99=A4=E8=BF=87=E6=9C=9F?= =?UTF-8?q?=E7=9A=84=E6=B5=8B=E8=AF=95=E4=BD=BF=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- endpoint/src/msgque/strategy/hitfirst.rs | 4 +- tests/src/ring_slice.rs | 60 ++++++++++++------------ 2 files changed, 32 insertions(+), 32 deletions(-) diff --git a/endpoint/src/msgque/strategy/hitfirst.rs b/endpoint/src/msgque/strategy/hitfirst.rs index 6307c0d60..5bb0ae061 100644 --- a/endpoint/src/msgque/strategy/hitfirst.rs +++ b/endpoint/src/msgque/strategy/hitfirst.rs @@ -152,7 +152,7 @@ impl HitFirstReader { // 此处说明已经读过 let last_read_idx = *lread_op.unwrap(); let mut found = 0; - for (i, ch) in self.cursors.iter().enumerate() { + for (_i, ch) in self.cursors.iter().enumerate() { let node_idx = ch.node_idx.load(Ordering::Relaxed); if node_idx == last_read_idx { let new_node_idx = if found == 0 { @@ -165,7 +165,7 @@ impl HitFirstReader { }; found += 1; - log::debug!("+++ shift cursor:{}, {} => {}", i, node_idx, new_node_idx); + log::debug!("+++ shift cursor:{}, {} => {}", _i, node_idx, new_node_idx); // last 读空，需要偏移位置 if let Ok(_c) = ch.node_idx.compare_exchange( node_idx, diff --git a/tests/src/ring_slice.rs b/tests/src/ring_slice.rs index 0b124d4f5..c4831e97a 100644 --- a/tests/src/ring_slice.rs +++ b/tests/src/ring_slice.rs @@ -93,34 +93,34 @@ mod tests_ds { // h //} - #[test] - fn test_split_ring_slice() { - println!("begin"); - - let data = "STORED\r\n"; - let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); - let index = slice.find_sub(0, "sdfsfdssssd".as_ref()); - println!("found = {}", index.is_some()); - - /* - let data = "END\r\nVALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\n"; - let slice = RingSlice::from(data.as_ptr(), data.len(), 21, data.len() + 21); - - //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND\r\n"; - //let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); - - //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND"; - //let slice = RingSlice::from(data.as_ptr(), 64, 0, data.len()); - println!("slice generated"); - let index = slice.find_sub(0, "\r\n".as_ref()); - println!("found, index = {}", index.unwrap()); - let split = slice.split("\r\n".as_ref()); - println!("slice split, size = {}", split.len()); - for single in split { - let mut single_vec: Vec = vec![]; - single.copy_to_vec(&mut single_vec); - println!("single = {}", String::from_utf8(single_vec).unwrap()); - } - */ - } + //#[test] + //fn test_split_ring_slice() { + // println!("begin"); + + // let data = "STORED\r\n"; + // let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); + // //let index = slice.find_sub(0, "sdfsfdssssd".as_ref()); + // //println!("found = {}", index.is_some()); + + // /* + // let data = "END\r\nVALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\n"; + // let slice = RingSlice::from(data.as_ptr(), data.len(), 21, data.len() + 21); + + // //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND\r\n"; + // //let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); + + // //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND"; + // //let slice = RingSlice::from(data.as_ptr(), 64, 0, data.len()); + // println!("slice generated"); + // let index = slice.find_sub(0, "\r\n".as_ref()); + // println!("found, index = {}", index.unwrap()); + // let split = slice.split("\r\n".as_ref()); + // println!("slice split, size = {}", split.len()); + // for single in split { + // let mut single_vec: Vec = vec![]; + // single.copy_to_vec(&mut single_vec); + // println!("single = {}", String::from_utf8(single_vec).unwrap()); + // } + // */ + //} } From 4accc80579b8a61d6e1793f144784fe4528b6acd Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 17:00:02 +0800 Subject: [PATCH 12/24] =?UTF-8?q?Enum=E7=9A=84layout=E5=9C=A8=E4=B8=8D?= =?UTF-8?q?=E5=90=8C=E7=9A=84=E7=8E=AF=E5=A2=83=E4=B8=8B=E4=B8=8D=E4=B8=80?= =?UTF-8?q?=E6=A0=B7=E3=80=82=E5=85=88=E6=B3=A8=E9=87=8A=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/src/layout.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/src/layout.rs b/tests/src/layout.rs index 167eb239d..50c8724ed 100644 --- a/tests/src/layout.rs +++ b/tests/src/layout.rs @@ -44,7 +44,7 @@ fn check_layout() { assert_eq!(handler_size, size_of::>()); assert_eq!(0, size_of::()); - assert_eq!(400, size_of::()); + //assert_eq!(400, size_of::()); assert_eq!(96, size_of::()); assert_eq!(264, size_of::()); assert_eq!(192, size_of::()); From dbe0f0c66f70286de71d5030b5f7689de271f3b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=B3=A2?= Date: Fri, 4 Nov 2022 17:48:22 +0800 Subject: [PATCH 13/24] =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E6=97=B6=EF=BC=8C=E4=B9=9F=E4=BF=AE=E6=94=B9ReconnPolicy?= =?UTF-8?q?=E7=9A=84conn=E8=AE=A1=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- stream/src/reconn.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/stream/src/reconn.rs b/stream/src/reconn.rs index 73b1d607d..988e8d962 100644 --- a/stream/src/reconn.rs +++ b/stream/src/reconn.rs @@ -21,6 +21,7 @@ impl ReconnPolicy { // 连接失败，为下一次连接做准备：sleep一段时间，避免无意义的重连 pub async fn conn_failed(&mut self) { + self.conns += 1; self.metric += 1; self.continue_fails += 1; From 03e28cc4b878f1994bb851e26073a2dbd8c5709e Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 18:59:20 +0800 Subject: [PATCH 14/24] =?UTF-8?q?=E5=9C=A8=E7=89=88=E6=9C=AC=E4=BF=A1?= =?UTF-8?q?=E6=81=AF=E4=B8=AD=E5=8A=A0=E5=85=A5debug=E6=A0=87=E8=AF=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- context/src/lib.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/context/src/lib.rs b/context/src/lib.rs index a552fa763..512679203 100644 --- a/context/src/lib.rs +++ b/context/src/lib.rs @@ -99,11 +99,12 @@ lazy_static! { let fields:Vec<&str> = full.split('-').collect(); let len = fields.len(); let last = *fields.get(len-1).unwrap_or(&""); + let build = if cfg!(debug_assertions) { "_debug" } else { "" }; if last == "modified" { let second_last = fields.get(len-2).unwrap_or(&""); - format!("{}_{}", second_last, last) + format!("{}_{}{}", second_last, last, build) } else { - last.to_string() + last.to_string() + build } }; static ref CONTEXT: Context = { From 54302c56ce56aed68bcc8cb3aa82b612ee8ed2ec Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 19:18:07 +0800 Subject: [PATCH 15/24] =?UTF-8?q?metrics:=20=E5=8E=BB=E6=8E=89pool=5Fname?= =?UTF-8?q?=20=E5=8E=BB=E6=8E=89ci=E4=B8=AD=E7=9A=84tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/rust.yml | 6 +++--- metrics/src/prometheus.rs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 2ef4f7d49..4d27787b5 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -54,7 +54,7 @@ jobs: ls -all ./target/debug/agent - name: Run run: nohup ./target/debug/agent --discovery vintage://127.0.0.1:8080 --snapshot /home/runner/work/breeze/snapshot --service-path /home/runner/work/breeze/socks --log-dir /home/runner/work/breeze/logs --port 9984 --metrics-probe 8.8.8.8:53 --log-level info > /home/runner/work/breeze/logs/log.file 2>&1 & - - name: Check Port + - name: Check Status run: | sleep 6s #netstat -nat|grep LISTEN @@ -71,5 +71,5 @@ jobs: with: version: '0.15.0' args: '-v' - - name: Run tests - run: cargo test + #- name: Run tests + # run: cargo test diff --git a/metrics/src/prometheus.rs b/metrics/src/prometheus.rs index 899bd28ef..3502759e1 100644 --- a/metrics/src/prometheus.rs +++ b/metrics/src/prometheus.rs @@ -177,7 +177,7 @@ impl<'a, 'r> ItemWriter for PrometheusItemWriter<'a, 'r> { self.first = true; //确保第一个put的label一定不为空; 后续优化 self.put_label("source", source); - self.put_label("pool", context::get().service_pool.as_bytes()); + //self.put_label("pool", context::get().service_pool.as_bytes()); self.put_label("namespace", namespace); self.put_label("topic", topic); self.put_label("bip", bip); From 36ef14a57079c6b9389ca78dc56dd80e944d8f57 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 19:27:28 +0800 Subject: [PATCH 16/24] =?UTF-8?q?metrics:=20=E9=83=A8=E5=88=86=E6=95=B4?= =?UTF-8?q?=E6=95=B0=E7=9A=84=E6=8C=87=E6=A0=87=E5=8E=BB=E6=8E=89=E5=B0=8F?= =?UTF-8?q?=E6=95=B0=E4=BD=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/src/types/rtt.rs | 6 +- metrics/src/types/tasks.rs | 6 -- tests/src/all.rs | 2 +- tests/src/layout.rs | 2 +- tests/src/queue.rs | 129 ++++++++++++++++++------------------- 5 files changed, 68 insertions(+), 77 deletions(-) delete mode 100644 metrics/src/types/tasks.rs diff --git a/metrics/src/types/rtt.rs b/metrics/src/types/rtt.rs index fd6eb383b..203e40418 100644 --- a/metrics/src/types/rtt.rs +++ b/metrics/src/types/rtt.rs @@ -19,12 +19,12 @@ impl Rtt { if count > 0 { w.write(&id.path, id.key, "qps", count as f64 / secs); // avg_us - let total_us = self.total_us.take() as f64; + let total_us = self.total_us.take(); // 按微秒取整 - let avg = (total_us / count as f64) as isize as f64; + let avg = (total_us as f64 / count as f64) as i64; w.write(&id.path, id.key, "avg_us", avg); - w.write(&id.path, id.key, "total_num", count as f64); + w.write(&id.path, id.key, "total_num", count); w.write(&id.path, id.key, "total_us", total_us); // slow qps let slow = self.slow.take(); diff --git a/metrics/src/types/tasks.rs b/metrics/src/types/tasks.rs deleted file mode 100644 index 91f4a4766..000000000 --- a/metrics/src/types/tasks.rs +++ /dev/null @@ -1,6 +0,0 @@ -use std::sync::atomic::{AtomicIsize, Ordering}; -pub static TASK_NUM: AtomicIsize = AtomicIsize::new(0); -pub(crate) fn snapshot(w: &mut W, _secs: f64) { - let num = TASK_NUM.load(Ordering::Relaxed); - w.write(BASE_PATH, "task", "num", num); -} diff --git a/tests/src/all.rs b/tests/src/all.rs index 066013972..c2b53b385 100644 --- a/tests/src/all.rs +++ b/tests/src/all.rs @@ -4,9 +4,9 @@ mod shard_test; //mod memcached_text; mod mem; -//mod queue; mod proto; mod protocol; +mod queue; // mod redis; mod hash_test; mod redis; diff --git a/tests/src/layout.rs b/tests/src/layout.rs index 50c8724ed..654e05385 100644 --- a/tests/src/layout.rs +++ b/tests/src/layout.rs @@ -7,7 +7,7 @@ use std::{ use protocol::Parser; use stream::{Backend, Request}; type Endpoint = Arc>; -type Topology = endpoint::Topology; +//type Topology = endpoint::Topology; type Stream = rt::Stream; diff --git a/tests/src/queue.rs b/tests/src/queue.rs index 2f3c74975..d4a8b7778 100644 --- a/tests/src/queue.rs +++ b/tests/src/queue.rs @@ -1,80 +1,77 @@ -#[cfg(test)] -mod queue { - use ds::PinnedQueue; - use std::sync::atomic::{AtomicIsize, Ordering::*}; +use ds::PinnedQueue; +use std::sync::atomic::{AtomicIsize, Ordering::*}; - static EXISTS: AtomicIsize = AtomicIsize::new(0); +static EXISTS: AtomicIsize = AtomicIsize::new(0); - struct Data { - v: u32, - } - impl Drop for Data { - fn drop(&mut self) { - EXISTS.fetch_sub(1, Relaxed); - } +struct Data { + v: u32, +} +impl Drop for Data { + fn drop(&mut self) { + EXISTS.fetch_sub(1, Relaxed); } +} - impl From for Data { - fn from(v: u32) -> Self { - EXISTS.fetch_add(1, Relaxed); - Self { v } - } +impl From for Data { + fn from(v: u32) -> Self { + EXISTS.fetch_add(1, Relaxed); + Self { v } } +} - #[test] - fn pinned_queue() { - let mut q: PinnedQueue = PinnedQueue::with_capacity(1); - q.push_back(0.into()); - assert_eq!(q.len(), 1); - q.pop_front().expect("pop front").v = 0; - assert_eq!(q.len(), 0); +#[test] +fn pinned_queue() { + let mut q: PinnedQueue = PinnedQueue::with_capacity(1); + q.push_back(0.into()); + assert_eq!(q.len(), 1); + q.pop_front().expect("pop front").v = 0; + assert_eq!(q.len(), 0); - q.push_back(1.into()); - q.push_back(2.into()); - q.push_back(3.into()); + q.push_back(1.into()); + q.push_back(2.into()); + q.push_back(3.into()); - assert_eq!(q.len(), 3); - unsafe { assert_eq!(q.pop_front_unchecked().v, 1) }; - unsafe { assert_eq!(q.pop_front_unchecked().v, 2) }; - unsafe { assert_eq!(q.pop_front_unchecked().v, 3) }; - assert!(q.pop_front().is_none()); - assert!(q.pop_front().is_none()); - assert!(q.pop_front().is_none()); + assert_eq!(q.len(), 3); + unsafe { assert_eq!(q.pop_front_unchecked().v, 1) }; + unsafe { assert_eq!(q.pop_front_unchecked().v, 2) }; + unsafe { assert_eq!(q.pop_front_unchecked().v, 3) }; + assert!(q.pop_front().is_none()); + assert!(q.pop_front().is_none()); + assert!(q.pop_front().is_none()); - assert_eq!(q.len(), 0); + assert_eq!(q.len(), 0); - // 随机验证。 - // 从0开始，随机写入m条，阿布读取n条。进行数据一致性验证。 - let MAX_CAP = 64; - let MAX_COUNT = 1000000; - // i 是插入过的数据条数量。 - // j 是pop出来的数据数量 - let mut i = 0; - let mut j = 0; - use rand::Rng; - let mut rng = rand::thread_rng(); - while i < MAX_COUNT { - let m = rng.gen::() % MAX_CAP; - for v in i..i + m { - q.push_back(v.into()); - } - i += m; - let n = rng.gen::() % MAX_CAP; - for _ in 0..n { - let v = q.pop_front(); - if j >= i { - assert_eq!(q.len(), 0); - assert!(v.is_none()); - } else { - assert_eq!(j, v.expect("take front").v); - j += 1; - } + // 随机验证。 + // 从0开始，随机写入m条，阿布读取n条。进行数据一致性验证。 + const MAX_CAP: u32 = 64; + const MAX_COUNT: u32 = 1000000; + // i 是插入过的数据条数量。 + // j 是pop出来的数据数量 + let mut i = 0; + let mut j = 0; + use rand::Rng; + let mut rng = rand::thread_rng(); + while i < MAX_COUNT { + let m = rng.gen::() % MAX_CAP; + for v in i..i + m { + q.push_back(v.into()); + } + i += m; + let n = rng.gen::() % MAX_CAP; + for _ in 0..n { + let v = q.pop_front(); + if j >= i { + assert_eq!(q.len(), 0); + assert!(v.is_none()); + } else { + assert_eq!(j, v.expect("take front").v); + j += 1; } } - - drop(q); - std::sync::atomic::fence(AcqRel); - let exists = EXISTS.load(Acquire); - assert_eq!(exists, 0); } + + drop(q); + std::sync::atomic::fence(AcqRel); + let exists = EXISTS.load(Acquire); + assert_eq!(exists, 0); } From 9f4b8c1e428f5a6dcb86ad48d442d115e56024bc Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 19:41:58 +0800 Subject: [PATCH 17/24] =?UTF-8?q?=E5=8F=96=E6=8E=89=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E7=9A=84=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- stream/src/handler.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/stream/src/handler.rs b/stream/src/handler.rs index a3009781c..a542c3fa1 100644 --- a/stream/src/handler.rs +++ b/stream/src/handler.rs @@ -71,7 +71,6 @@ where fn poll_request(&mut self, cx: &mut Context) -> Poll> { self.s.cache(self.data.size_hint() > 1); while let Some(req) = ready!(self.data.poll_recv(cx)) { - log::info!("request:{:?}", req); self.num_tx += 1; self.s.write_slice(req.data(), 0)?; match req.on_sent() { @@ -98,7 +97,6 @@ where self.num_rx += 1; // 统计请求耗时。 self.rtt += req.start_at().elapsed(); - log::info!("response: {:?}", cmd); req.on_complete(cmd); } } From f33575eed048079ad1cd7ba8b14b7d9571e6f700 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Fri, 4 Nov 2022 21:07:22 +0800 Subject: [PATCH 18/24] =?UTF-8?q?metrics:=20=E5=8E=BB=E5=B9=B4HELP?= =?UTF-8?q?=E4=BF=A1=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/src/prometheus.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/metrics/src/prometheus.rs b/metrics/src/prometheus.rs index 3502759e1..1f49752a8 100644 --- a/metrics/src/prometheus.rs +++ b/metrics/src/prometheus.rs @@ -162,9 +162,9 @@ impl<'a, 'r> ItemWriter for PrometheusItemWriter<'a, 'r> { let metric_name = MetricName(key, sub_key); //promethues # HELP - self.put_slice("# HELP "); - metric_name.write_to(self); - self.put_slice("\n"); + //self.put_slice("# HELP "); + //metric_name.write_to(self); + //self.put_slice("\n"); //promethues # TYPE self.put_slice("# TYPE "); From 8dff92da7bfa964c8472a6b50699b71c8e872d95 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=B3=A2?= Date: Sat, 5 Nov 2022 09:45:19 +0800 Subject: [PATCH 19/24] =?UTF-8?q?=E7=BB=8F=E9=AA=8C=E8=AF=81=EF=BC=8Ccrc32?= =?UTF-8?q?=E3=80=81crc32local=E7=9A=84feed=20tab=E7=9B=B8=E5=90=8C?= =?UTF-8?q?=EF=BC=8C=E5=8F=AF=E5=A4=8D=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sharding/src/hash/crc32.rs | 4 +-- sharding/src/hash/crc32local.rs | 54 ++++++--------------------------- 2 files changed, 12 insertions(+), 46 deletions(-) diff --git a/sharding/src/hash/crc32.rs b/sharding/src/hash/crc32.rs index 8bf99c263..c77f37bd3 100644 --- a/sharding/src/hash/crc32.rs +++ b/sharding/src/hash/crc32.rs @@ -1,6 +1,6 @@ use std::fmt::Display; -const CRC32TAB: [i64; 256] = [ +pub(super) const CRC32TAB: [i64; 256] = [ 0x00000000, 0x77073096, 0xEE0E612C, 0x990951BA, 0x076DC419, 0x706AF48F, 0xE963A535, 0x9E6495A3, 0x0EDB8832, 0x79DCB8A4, 0xE0D5E91E, 0x97D2D988, 0x09B64C2B, 0x7EB17CBD, 0xE7B82D07, 0x90BF1D91, 0x1DB71064, 0x6AB020F2, 0xF3B97148, 0x84BE41DE, 0x1ADAD47D, 0x6DDDE4EB, 0xF4D4B551, 0x83D385C7, @@ -35,7 +35,7 @@ const CRC32TAB: [i64; 256] = [ 0xB3667A2E, 0xC4614AB8, 0x5D681B02, 0x2A6F2B94, 0xB40BBE37, 0xC30C8EA1, 0x5A05DF1B, 0x2D02EF8D, ]; -const CRC_SEED: i64 = 0xFFFFFFFF; +pub(super) const CRC_SEED: i64 = 0xFFFFFFFF; // 用于兼容jdk版本crc32算法 #[derive(Default, Clone, Debug)] diff --git a/sharding/src/hash/crc32local.rs b/sharding/src/hash/crc32local.rs index 9cd2787bc..08ab5afa7 100644 --- a/sharding/src/hash/crc32local.rs +++ b/sharding/src/hash/crc32local.rs @@ -2,58 +2,24 @@ use std::fmt::Display; -use super::Hash; - -const CRCLOCALTAB: [i64; 256] = [ - 0x00000000, 0x77073096, 0xee0e612c, 0x990951ba, 0x076dc419, 0x706af48f, 0xe963a535, 0x9e6495a3, - 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988, 0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91, - 0x1db71064, 0x6ab020f2, 0xf3b97148, 0x84be41de, 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7, - 0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec, 0x14015c4f, 0x63066cd9, 0xfa0f3d63, 0x8d080df5, - 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172, 0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b, - 0x35b5a8fa, 0x42b2986c, 0xdbbbc9d6, 0xacbcf940, 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59, - 0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116, 0x21b4f4b5, 0x56b3c423, 0xcfba9599, 0xb8bda50f, - 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924, 0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d, - 0x76dc4190, 0x01db7106, 0x98d220bc, 0xefd5102a, 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433, - 0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818, 0x7f6a0dbb, 0x086d3d2d, 0x91646c97, 0xe6635c01, - 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e, 0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457, - 0x65b0d9c6, 0x12b7e950, 0x8bbeb8ea, 0xfcb9887c, 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65, - 0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2, 0x4adfa541, 0x3dd895d7, 0xa4d1c46d, 0xd3d6f4fb, - 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0, 0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9, - 0x5005713c, 0x270241aa, 0xbe0b1010, 0xc90c2086, 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f, - 0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4, 0x59b33d17, 0x2eb40d81, 0xb7bd5c3b, 0xc0ba6cad, - 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a, 0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683, - 0xe3630b12, 0x94643b84, 0x0d6d6a3e, 0x7a6a5aa8, 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1, - 0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe, 0xf762575d, 0x806567cb, 0x196c3671, 0x6e6b06e7, - 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc, 0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5, - 0xd6d6a3e8, 0xa1d1937e, 0x38d8c2c4, 0x4fdff252, 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b, - 0xd80d2bda, 0xaf0a1b4c, 0x36034af6, 0x41047a60, 0xdf60efc3, 0xa867df55, 0x316e8eef, 0x4669be79, - 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236, 0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f, - 0xc5ba3bbe, 0xb2bd0b28, 0x2bb45a92, 0x5cb36a04, 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d, - 0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a, 0x9c0906a9, 0xeb0e363f, 0x72076785, 0x05005713, - 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38, 0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21, - 0x86d3d2d4, 0xf1d4e242, 0x68ddb3f8, 0x1fda836e, 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777, - 0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c, 0x8f659eff, 0xf862ae69, 0x616bffd3, 0x166ccf45, - 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2, 0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db, - 0xaed16a4a, 0xd9d65adc, 0x40df0b66, 0x37d83bf0, 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9, - 0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6, 0xbad03605, 0xcdd70693, 0x54de5729, 0x23d967bf, - 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94, 0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d, -]; - -const CRCLOCAL_SEED: i64 = 0xffffffff; +use super::{ + crc32::{CRC32TAB, CRC_SEED}, + Hash, +}; #[derive(Default, Clone, Debug)] pub struct Crc32local {} impl Hash for Crc32local { fn hash(&self, key: &S) -> i64 { - let mut crc: i64 = CRCLOCAL_SEED; + let mut crc: i64 = CRC_SEED; for i in 0..key.len() { let c = key.at(i); - crc = crc >> 8 ^ CRCLOCALTAB[((crc ^ (c as i64)) & 0xff) as usize]; + crc = crc >> 8 ^ CRC32TAB[((crc ^ (c as i64)) & 0xff) as usize]; } - crc ^= CRCLOCAL_SEED; + crc ^= CRC_SEED; let crc32 = crc as i32; crc32.abs() as i64 } @@ -104,7 +70,7 @@ impl Crc32localDelimiter { impl super::Hash for Crc32localDelimiter { fn hash(&self, key: &S) -> i64 { - let mut crc: i64 = CRCLOCAL_SEED; + let mut crc: i64 = CRC_SEED; debug_assert!(self.start_pos < key.len()); // 对于用“.”、“_”、“#”做分割的hash key，遇到分隔符停止 @@ -114,10 +80,10 @@ impl super::Hash for Crc32localDelimiter { if check_delimiter && (c == self.delimiter) { break; } - crc = crc >> 8 ^ CRCLOCALTAB[((crc ^ (c as i64)) & 0xff) as usize]; + crc = crc >> 8 ^ CRC32TAB[((crc ^ (c as i64)) & 0xff) as usize]; } - crc ^= CRCLOCAL_SEED; + crc ^= CRC_SEED; let crc32 = crc as i32; crc32.abs() as i64 From 5aa15dd1b0f6e75a5fd0f340a69535e15fbae942 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Sat, 5 Nov 2022 12:36:58 +0800 Subject: [PATCH 20/24] =?UTF-8?q?tests:=20ringlice=E9=9A=8F=E6=9C=BA?= =?UTF-8?q?=E7=94=9F=E6=88=90=E5=AD=97=E7=AC=A6=E4=B8=B2=EF=BC=8C=E9=99=90?= =?UTF-8?q?=E5=AE=9A=E5=B0=8F=E5=86=99=E5=AD=97=E6=AF=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/src/ring_slice.rs | 247 ++++++++++++++++++++-------------------- 1 file changed, 123 insertions(+), 124 deletions(-) diff --git a/tests/src/ring_slice.rs b/tests/src/ring_slice.rs index c4831e97a..03ddb65d0 100644 --- a/tests/src/ring_slice.rs +++ b/tests/src/ring_slice.rs @@ -1,126 +1,125 @@ -#[cfg(test)] -mod tests_ds { - use ds::RingSlice; - #[test] - fn test_ring_slice() { - let cap = 1024; - let mut data: Vec = (0..cap).map(|_| rand::random::()).collect(); - let dc = data.clone(); - let ptr = data.as_mut_ptr(); - std::mem::forget(data); - - let in_range = RingSlice::from(ptr, cap, 0, 32); - assert_eq!(in_range.len(), 32); - assert_eq!(in_range.read(0), &dc[0..32]); - - // 截止到末尾的 - let end_range = RingSlice::from(ptr, cap, cap - 32, cap); - //let s = end_range.as_slices(); - //assert_eq!(s.len(), 1); - assert_eq!(end_range.len(), 32); - assert_eq!(end_range.read(0), &dc[cap - 32..cap]); - - let over_range = RingSlice::from(ptr, cap, cap - 32, cap + 32); - //let s = over_range.as_slices(); - //assert_eq!(over_range.len(), 64); - //assert_eq!(s.len(), 2); - assert_eq!(over_range.read(0), &dc[cap - 32..cap]); - assert_eq!(over_range.read(32), &dc[0..32]); - let mut v: Vec = Vec::new(); - over_range.copy_to_vec(&mut v); - assert_eq!(&v[0..32], &dc[cap - 32..cap]); - assert_eq!(&v[32..], &dc[0..32]); - - let u32_num = 111234567u32; - let bytes = u32_num.to_be_bytes(); - unsafe { - //log::debug!("bytes:{:?}", bytes); - std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr.offset(8), 4); - std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr.offset(1023), 1); - std::ptr::copy_nonoverlapping(bytes.as_ptr().offset(1), ptr, 3); - } - let num_range = RingSlice::from(ptr, cap, 1000, 1064); - assert_eq!(u32_num, num_range.read_u32(32)); - - assert_eq!(u32_num, num_range.read_u32(23)); - - // 验证查找\r\n - let mut lines = RingSlice::from(ptr, cap, cap - 32, cap + 32); - lines.update(9, b'\r'); - lines.update(20, b'\r'); - lines.update(21, b'\n'); - lines.update(62, b'\r'); - lines.update(63, b'\n'); - - let line = lines.find_lf_cr(0); - let r = lines.find(0, b'\r'); - assert!(line.is_some()); - assert!(r.is_some()); - assert_eq!(line.expect("line"), 20); - assert_eq!(r.expect("line-r"), 9); - assert_eq!(lines.find_lf_cr(22).unwrap(), 62); - - let _ = unsafe { Vec::from_raw_parts(ptr, 0, cap) }; +use ds::RingSlice; +#[test] +fn test_ring_slice() { + let cap = 1024; + let mut data: Vec = (0..cap) + .map(|_| rand::random::().max(b'a').min(b'z')) + .collect(); + let dc = data.clone(); + let ptr = data.as_mut_ptr(); + std::mem::forget(data); + + let in_range = RingSlice::from(ptr, cap, 0, 32); + assert_eq!(in_range.len(), 32); + assert_eq!(in_range.read(0), &dc[0..32]); + + // 截止到末尾的 + let end_range = RingSlice::from(ptr, cap, cap - 32, cap); + //let s = end_range.as_slices(); + //assert_eq!(s.len(), 1); + assert_eq!(end_range.len(), 32); + assert_eq!(end_range.read(0), &dc[cap - 32..cap]); + + let over_range = RingSlice::from(ptr, cap, cap - 32, cap + 32); + //let s = over_range.as_slices(); + //assert_eq!(over_range.len(), 64); + //assert_eq!(s.len(), 2); + assert_eq!(over_range.read(0), &dc[cap - 32..cap]); + assert_eq!(over_range.read(32), &dc[0..32]); + let mut v: Vec = Vec::new(); + over_range.copy_to_vec(&mut v); + assert_eq!(&v[0..32], &dc[cap - 32..cap]); + assert_eq!(&v[32..], &dc[0..32]); + + let u32_num = 111234567u32; + let bytes = u32_num.to_be_bytes(); + unsafe { + //log::debug!("bytes:{:?}", bytes); + std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr.offset(8), 4); + std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr.offset(1023), 1); + std::ptr::copy_nonoverlapping(bytes.as_ptr().offset(1), ptr, 3); } - //#[test] - //fn test_ring_slice_map() { - // let slice_data = [98u8, 114, 101, 101, 122, 101, 45, 107, 101, 121, 45, 56, 57]; - // let slice = Slice::from(&slice_data); - - // // 第一个字节在最后 - // let ring_slice_data: [u8; 16] = [ - // 114, 101, 101, 122, 101, 45, 107, 101, 121, 45, 56, 57, 0, 0, 0, 98, - // ]; - // let ring_slice = RingSlice::from(ring_slice_data.as_ptr(), 16, 15, 15 + slice_data.len()); - - // assert_eq!(ring_slice, slice); - // let slice_to_ring: RingSlice = slice.clone().into(); - // assert_eq!(ring_slice, slice_to_ring); - - // //assert_eq!(hash(&slice_to_ring), hash(&ring_slice)); - - // let mut m = HashMap::with_capacity(4); - // m.insert(ring_slice, ()); - // assert!(m.contains_key(&slice.into())); - //} - //use std::collections::hash_map::DefaultHasher; - //use std::hash::{Hash, Hasher}; - //fn hash(t: &T) -> u64 { - // let mut s = DefaultHasher::new(); - // t.hash(&mut s); - // let h = s.finish(); - // println!("hash ring:{}", h); - // h - //} - - //#[test] - //fn test_split_ring_slice() { - // println!("begin"); - - // let data = "STORED\r\n"; - // let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); - // //let index = slice.find_sub(0, "sdfsfdssssd".as_ref()); - // //println!("found = {}", index.is_some()); - - // /* - // let data = "END\r\nVALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\n"; - // let slice = RingSlice::from(data.as_ptr(), data.len(), 21, data.len() + 21); - - // //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND\r\n"; - // //let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); - - // //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND"; - // //let slice = RingSlice::from(data.as_ptr(), 64, 0, data.len()); - // println!("slice generated"); - // let index = slice.find_sub(0, "\r\n".as_ref()); - // println!("found, index = {}", index.unwrap()); - // let split = slice.split("\r\n".as_ref()); - // println!("slice split, size = {}", split.len()); - // for single in split { - // let mut single_vec: Vec = vec![]; - // single.copy_to_vec(&mut single_vec); - // println!("single = {}", String::from_utf8(single_vec).unwrap()); - // } - // */ - //} + let num_range = RingSlice::from(ptr, cap, 1000, 1064); + assert_eq!(u32_num, num_range.read_u32(32)); + + assert_eq!(u32_num, num_range.read_u32(23)); + + // 验证查找\r\n + let mut lines = RingSlice::from(ptr, cap, cap - 32, cap + 32); + lines.update(9, b'\r'); + lines.update(20, b'\r'); + lines.update(21, b'\n'); + lines.update(62, b'\r'); + lines.update(63, b'\n'); + + let line = lines.find_lf_cr(0); + let r = lines.find(0, b'\r'); + assert!(line.is_some()); + assert!(r.is_some()); + assert_eq!(line.expect("line"), 20); + assert_eq!(r.expect("line-r"), 9); + assert_eq!(lines.find_lf_cr(22).unwrap(), 62); + + let _ = unsafe { Vec::from_raw_parts(ptr, 0, cap) }; } +//#[test] +//fn test_ring_slice_map() { +// let slice_data = [98u8, 114, 101, 101, 122, 101, 45, 107, 101, 121, 45, 56, 57]; +// let slice = Slice::from(&slice_data); + +// // 第一个字节在最后 +// let ring_slice_data: [u8; 16] = [ +// 114, 101, 101, 122, 101, 45, 107, 101, 121, 45, 56, 57, 0, 0, 0, 98, +// ]; +// let ring_slice = RingSlice::from(ring_slice_data.as_ptr(), 16, 15, 15 + slice_data.len()); + +// assert_eq!(ring_slice, slice); +// let slice_to_ring: RingSlice = slice.clone().into(); +// assert_eq!(ring_slice, slice_to_ring); + +// //assert_eq!(hash(&slice_to_ring), hash(&ring_slice)); + +// let mut m = HashMap::with_capacity(4); +// m.insert(ring_slice, ()); +// assert!(m.contains_key(&slice.into())); +//} +//use std::collections::hash_map::DefaultHasher; +//use std::hash::{Hash, Hasher}; +//fn hash(t: &T) -> u64 { +// let mut s = DefaultHasher::new(); +// t.hash(&mut s); +// let h = s.finish(); +// println!("hash ring:{}", h); +// h +//} + +//#[test] +//fn test_split_ring_slice() { +// println!("begin"); + +// let data = "STORED\r\n"; +// let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); +// //let index = slice.find_sub(0, "sdfsfdssssd".as_ref()); +// //println!("found = {}", index.is_some()); + +// /* +// let data = "END\r\nVALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\n"; +// let slice = RingSlice::from(data.as_ptr(), data.len(), 21, data.len() + 21); + +// //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND\r\n"; +// //let slice = RingSlice::from(data.as_ptr(), data.len(), 0, data.len()); + +// //let data = "VALUE key1 0 9\r\nssksksksk\r\nVALUE key2 0 13\r\nabababababaab\r\nEND"; +// //let slice = RingSlice::from(data.as_ptr(), 64, 0, data.len()); +// println!("slice generated"); +// let index = slice.find_sub(0, "\r\n".as_ref()); +// println!("found, index = {}", index.unwrap()); +// let split = slice.split("\r\n".as_ref()); +// println!("slice split, size = {}", split.len()); +// for single in split { +// let mut single_vec: Vec = vec![]; +// single.copy_to_vec(&mut single_vec); +// println!("single = {}", String::from_utf8(single_vec).unwrap()); +// } +// */ +//} From 9cc611ecc5b7740e0126c752dda9ce717d88c169 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Sun, 6 Nov 2022 16:14:10 +0800 Subject: [PATCH 21/24] =?UTF-8?q?heap=20stats:=20=E5=A2=9E=E5=8A=A0heap?= =?UTF-8?q?=E7=9A=84=E6=8C=87=E6=A0=87=E9=87=87=E9=9B=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent/Cargo.toml | 1 - agent/src/main.rs | 4 +-- ds/Cargo.toml | 6 ++++ ds/src/mem/malloc.rs | 67 +++++++++++++++++++++++++++++++++++++++ ds/src/mem/mod.rs | 3 ++ metrics/src/types/host.rs | 30 ++++++++++++++---- 6 files changed, 101 insertions(+), 10 deletions(-) create mode 100644 ds/src/mem/malloc.rs diff --git a/agent/Cargo.toml b/agent/Cargo.toml index b1fc16fbc..732513c09 100644 --- a/agent/Cargo.toml +++ b/agent/Cargo.toml @@ -25,7 +25,6 @@ backtrace = "0.3.63" lazy_static = "1.4.0" tokio = { version = "1.21.2", features = ["rt", "net", "rt-multi-thread", "time"], default-features = false } -mimalloc = { version = "*", default-features = false } once_cell = "*" rlimit = "0.8.3" diff --git a/agent/src/main.rs b/agent/src/main.rs index 1926327db..d6939422b 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -1,6 +1,6 @@ -use mimalloc::MiMalloc; +use ds::BrzMalloc; #[global_allocator] -static GLOBAL: MiMalloc = MiMalloc; +static GLOBAL: BrzMalloc = BrzMalloc {}; #[macro_use] extern crate rocket; diff --git a/ds/Cargo.toml b/ds/Cargo.toml index 235731a4a..76b3186b6 100644 --- a/ds/Cargo.toml +++ b/ds/Cargo.toml @@ -12,3 +12,9 @@ tokio = { version = "1.21.2", features = ["rt", "net", "rt-multi-thread", "time" byteorder = "1.4.3" rand = "0.8.4" atomic-waker = "*" +mimalloc = { version = "*", default-features = false } +cache-padded = "1.2.0" + +[features] +default = ["heap-stats"] +heap-stats = [] diff --git a/ds/src/mem/malloc.rs b/ds/src/mem/malloc.rs new file mode 100644 index 000000000..4d8ed7ca6 --- /dev/null +++ b/ds/src/mem/malloc.rs @@ -0,0 +1,67 @@ +pub struct HeapStats { + pub total: usize, + pub used: usize, + pub total_objects: usize, + pub used_objects: usize, +} +pub use inner::*; +#[cfg(feature = "heap-stats")] +mod inner { + use cache_padded::CachePadded; + use mimalloc::MiMalloc; + use std::alloc::{GlobalAlloc, Layout}; + use std::sync::atomic::{AtomicU64, Ordering::*}; + static ALLOC: CachePadded = CachePadded::new(AtomicU64::new(0)); + static FREE: CachePadded = CachePadded::new(AtomicU64::new(0)); + static ALLOC_OBJ: CachePadded = CachePadded::new(AtomicU64::new(0)); + static FREE_OBJ: CachePadded = CachePadded::new(AtomicU64::new(0)); + + struct Stats; + impl Stats { + #[inline(always)] + fn alloc(&self, size: usize) { + ALLOC.fetch_add(size as u64, Relaxed); + ALLOC_OBJ.fetch_add(1, Relaxed); + } + #[inline(always)] + fn free(&self, size: usize) { + FREE.fetch_add(size as u64, Relaxed); + FREE_OBJ.fetch_add(1, Relaxed); + } + } + pub struct BrzMalloc; + unsafe impl GlobalAlloc for BrzMalloc { + #[inline] + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + Stats.alloc(layout.size()); + MiMalloc.alloc(layout) + } + + #[inline] + unsafe fn dealloc(&self, ptr: *mut u8, _layout: Layout) { + Stats.free(_layout.size()); + MiMalloc.dealloc(ptr, _layout) + } + } + + pub fn heap() -> Option { + let alloc = ALLOC.load(Relaxed); + let free = FREE.load(Relaxed); + let alloc_objects = ALLOC_OBJ.load(Relaxed); + let free_objects = FREE_OBJ.load(Relaxed); + Some(super::HeapStats { + total: alloc as usize, + used: (alloc - free) as usize, + total_objects: alloc_objects as usize, + used_objects: (alloc_objects - free_objects) as usize, + }) + } +} +#[cfg(not(feature = "heap-stats"))] +mod inner { + pub struct HeapStats; + pub type BrzMalloc = mimalloc::MiMalloc; + pub fn heap() -> Option { + None + } +} diff --git a/ds/src/mem/mod.rs b/ds/src/mem/mod.rs index c6415814a..6d4288919 100644 --- a/ds/src/mem/mod.rs +++ b/ds/src/mem/mod.rs @@ -12,3 +12,6 @@ pub use guarded::*; mod policy; pub use policy::*; + +mod malloc; +pub use malloc::*; diff --git a/metrics/src/types/host.rs b/metrics/src/types/host.rs index 7f0748ebb..a009a4560 100644 --- a/metrics/src/types/host.rs +++ b/metrics/src/types/host.rs @@ -11,6 +11,7 @@ static TASK_NUM: AtomicI64 = AtomicI64::new(0); static SOCKFILE_FAILED: AtomicI64 = AtomicI64::new(0); pub struct Host { + heap: Option, // 累积分配的堆内存 start: Instant, process: Process, version: &'static str, @@ -20,23 +21,21 @@ impl Host { #[inline] pub(crate) fn new() -> Self { Self { + heap: None, start: Instant::now(), process: Process::current().expect("cannot get current process"), version: &context::get().version, } } #[inline] - pub(crate) fn snapshot(&mut self, w: &mut W, _secs: f64) { + pub(crate) fn snapshot(&mut self, w: &mut W, secs: f64) { self.refresh(); - w.write( - BASE_PATH, - "host", - "uptime_sec", - self.start.elapsed().as_secs() as i64, - ); + let uptime = self.start.elapsed().as_secs() as i64; + w.write(BASE_PATH, "host", "uptime_sec", uptime); let percent = CPU_PERCENT.load(Ordering::Relaxed) as f64 / 100.0; w.write(BASE_PATH, "host", "cpu", percent); w.write(BASE_PATH, "host", "mem", MEMORY.load(Ordering::Relaxed)); + self.snapshot_heap(w, secs); let tasks = TASK_NUM.load(Ordering::Relaxed); w.write(BASE_PATH, "task", "num", tasks); @@ -45,6 +44,23 @@ impl Host { let sockfile_failed = SOCKFILE_FAILED.load(Ordering::Relaxed); w.write(BASE_PATH, "sockfile", "failed", sockfile_failed); } + pub(crate) fn snapshot_heap(&mut self, w: &mut W, _secs: f64) { + if let Some(heap_stats) = ds::heap() { + // 已使用堆内存 + w.write(BASE_PATH, "host", "heap", heap_stats.used as i64); + // 已分配的对象的数量 + w.write(BASE_PATH, "host", "heap_o", heap_stats.used_objects as i64); + if let Some(prev) = self.heap.take() { + // 堆内存分配速率 + let bps = ((heap_stats.total - prev.total) as f64 / _secs) as i64; + w.write(BASE_PATH, "host", "heap_bps", bps); + // 堆分配对象速率 + let ops = ((heap_stats.total_objects - prev.total_objects) as f64 / _secs) as i64; + w.write(BASE_PATH, "host", "heap_ops", ops); + } + self.heap = Some(heap_stats); + } + } pub(crate) fn refresh(&mut self) { if let Ok(percent) = self.process.cpu_percent() { CPU_PERCENT.store((percent * 100.0) as usize, Ordering::Relaxed); From bddaa77906579b57c7916b03b0a2f91fcf5ae6cc Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Sun, 6 Nov 2022 20:37:45 +0800 Subject: [PATCH 22/24] =?UTF-8?q?time:=20=E8=87=AA=E5=AE=9A=E4=B9=89ds::ti?= =?UTF-8?q?me=EF=BC=8C=E5=90=8E=E7=BB=AD=E6=9B=BF=E6=8D=A2std::time?= =?UTF-8?q?=E3=80=82=E9=BB=98=E8=AE=A4=E4=BD=BF=E7=94=A8std::time?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ds/src/lib.rs | 2 ++ ds/src/time/mod.rs | 2 ++ 2 files changed, 4 insertions(+) create mode 100644 ds/src/time/mod.rs diff --git a/ds/src/lib.rs b/ds/src/lib.rs index 54d5b910c..3ea895794 100644 --- a/ds/src/lib.rs +++ b/ds/src/lib.rs @@ -17,6 +17,8 @@ pub use switcher::Switcher; pub use utf8::*; pub use waker::AtomicWaker; +pub mod time; + pub const NUM_STR_TBL: [&'static str; 32] = [ "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "30", "31", diff --git a/ds/src/time/mod.rs b/ds/src/time/mod.rs new file mode 100644 index 000000000..646d99033 --- /dev/null +++ b/ds/src/time/mod.rs @@ -0,0 +1,2 @@ +pub type Instant = std::time::Instant; +pub type Duration = std::time::Duration; From aa0185282846afb14bdab44e63cfb4ac289625ee Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Sun, 6 Nov 2022 20:44:52 +0800 Subject: [PATCH 23/24] =?UTF-8?q?ds::asserts=EF=BC=8C=E5=B0=86asserts?= =?UTF-8?q?=E7=94=B1ds=E6=9D=A5=E4=BB=A3=E7=90=86=EF=BC=8C=E5=90=8E?= =?UTF-8?q?=E7=BB=AD=E6=96=B9=E4=BE=BF=E7=BB=9F=E4=B8=80=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ds/src/asserts.rs | 10 ++++++++++ ds/src/lib.rs | 3 +++ 2 files changed, 13 insertions(+) create mode 100644 ds/src/asserts.rs diff --git a/ds/src/asserts.rs b/ds/src/asserts.rs new file mode 100644 index 000000000..be9e46e57 --- /dev/null +++ b/ds/src/asserts.rs @@ -0,0 +1,10 @@ +#[macro_export] +macro_rules! assert { + ($cond:expr $(,)?) => {{ std::assert!($cond $(,)?) }}; + ($cond:expr, $($arg:tt)+) => {{ std::assert!($cond, $($arg)+) }}; +} +#[macro_export] +macro_rules! assert_eq { + ($cond:expr $(,)?) => {{ std::assert!($cond $(,)?) }}; + ($cond:expr, $($arg:tt)+) => {{ std::assert!($cond, $($arg)+) }}; +} diff --git a/ds/src/lib.rs b/ds/src/lib.rs index 3ea895794..cf994dd76 100644 --- a/ds/src/lib.rs +++ b/ds/src/lib.rs @@ -19,6 +19,9 @@ pub use waker::AtomicWaker; pub mod time; +mod asserts; +pub use asserts::*; + pub const NUM_STR_TBL: [&'static str; 32] = [ "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "30", "31", From 29e7819e764b9395836ae11df7a71388209556f6 Mon Sep 17 00:00:00 2001 From: icycrystal4 Date: Sun, 6 Nov 2022 20:59:38 +0800 Subject: [PATCH 24/24] =?UTF-8?q?time:=20=E8=87=AA=E5=AE=9A=E4=B9=89ds::ti?= =?UTF-8?q?me=EF=BC=8C=E5=90=8E=E7=BB=AD=E6=9B=BF=E6=8D=A2std::time?= =?UTF-8?q?=E3=80=82=E9=BB=98=E8=AE=A4=E4=BD=BF=E7=94=A8std::time?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent/src/main.rs | 2 +- agent/src/prometheus.rs | 4 ++-- agent/src/service.rs | 2 +- context/Cargo.toml | 1 + context/src/lib.rs | 4 ++-- context/src/quadruple.rs | 2 +- discovery/src/cfg.rs | 2 +- discovery/src/dns.rs | 2 +- discovery/src/update.rs | 2 +- ds/src/mem/policy.rs | 4 ++-- endpoint/src/cacheservice/config.rs | 2 +- endpoint/src/cacheservice/topo.rs | 2 +- endpoint/src/lib.rs | 2 +- endpoint/src/phantomservice/config.rs | 2 +- endpoint/src/redisservice/config.rs | 2 +- endpoint/src/redisservice/topo.rs | 2 +- metrics/src/macros.rs | 2 +- metrics/src/register.rs | 2 +- metrics/src/sender.rs | 2 +- metrics/src/types/host.rs | 2 +- metrics/src/types/rtt.rs | 2 +- protocol/src/callback.rs | 2 +- protocol/src/error.rs | 2 +- protocol/src/lib.rs | 2 +- protocol/src/req.rs | 2 +- protocol/src/request.rs | 2 +- rt/src/entry.rs | 2 +- stream/src/builder.rs | 2 +- stream/src/checker.rs | 2 +- stream/src/pipeline.rs | 2 +- stream/src/reconn.rs | 2 +- tests/src/asserts.rs | 4 ++++ tests/src/mem.rs | 2 +- tests/src/ring_buffer.rs | 6 +++--- 34 files changed, 42 insertions(+), 37 deletions(-) create mode 100644 tests/src/asserts.rs diff --git a/agent/src/main.rs b/agent/src/main.rs index d6939422b..948f0f734 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -13,7 +13,7 @@ use discovery::*; mod init; use rt::spawn; -use std::time::Duration; +use ds::time::Duration; use protocol::Result; diff --git a/agent/src/prometheus.rs b/agent/src/prometheus.rs index 80f083a5d..19218393d 100644 --- a/agent/src/prometheus.rs +++ b/agent/src/prometheus.rs @@ -19,7 +19,7 @@ pub struct PrometheusMetricsResponse {} use ds::lock::Lock; use lazy_static::lazy_static; -use std::time::Instant; +use ds::time::Instant; lazy_static! { static ref LAST: Lock = Instant::now().into(); @@ -71,7 +71,7 @@ pub(crate) fn register_target(ctx: &context::Context) { port ); let client = reqwest::Client::new(); - let mut interval = tokio::time::interval(std::time::Duration::from_secs(60)); + let mut interval = tokio::time::interval(ds::time::Duration::from_secs(60)); let mut q = vec![("refresh", true)]; loop { let body = body.clone(); diff --git a/agent/src/service.rs b/agent/src/service.rs index 62a74314d..38d6479c3 100644 --- a/agent/src/service.rs +++ b/agent/src/service.rs @@ -2,7 +2,7 @@ use context::Quadruple; use net::Listener; use rt::spawn; use std::sync::Arc; -use std::time::Duration; +use ds::time::Duration; use discovery::TopologyWriteGuard; use ds::chan::Sender; diff --git a/context/Cargo.toml b/context/Cargo.toml index 87d9a267c..0a64f621b 100644 --- a/context/Cargo.toml +++ b/context/Cargo.toml @@ -8,6 +8,7 @@ edition = "2021" [dependencies] log = { path = "../log" } +ds = { path = "../ds" } clap = {version = "3.1.1", features = ["derive"] } url = "2.2.2" diff --git a/context/src/lib.rs b/context/src/lib.rs index 512679203..5f14e9234 100644 --- a/context/src/lib.rs +++ b/context/src/lib.rs @@ -134,9 +134,9 @@ impl ContextOption { Ok(()) } - pub fn tick(&self) -> std::time::Duration { + pub fn tick(&self) -> ds::time::Duration { assert!(self.tick_sec >= 1 && self.tick_sec <= 60); - std::time::Duration::from_secs(self.tick_sec as u64) + ds::time::Duration::from_secs(self.tick_sec as u64) } // 如果是以升级模式启动，则会将原有的端口先关闭。 pub fn listeners(&self) -> ListenerIter { diff --git a/context/src/quadruple.rs b/context/src/quadruple.rs index 12c741076..00a9128d5 100644 --- a/context/src/quadruple.rs +++ b/context/src/quadruple.rs @@ -1,5 +1,5 @@ use std::path::Path; -use std::time::{Duration, Instant}; +use ds::time::{Duration, Instant}; #[derive(Debug, Clone, Eq)] pub struct Quadruple { parsed_at: Instant, diff --git a/discovery/src/cfg.rs b/discovery/src/cfg.rs index 07422397e..fb9396d00 100644 --- a/discovery/src/cfg.rs +++ b/discovery/src/cfg.rs @@ -1,6 +1,6 @@ use std::io::{Error, ErrorKind, Result}; use std::path::PathBuf; -use std::time::{Duration, Instant}; +use ds::time::{Duration, Instant}; use tokio::fs::File; use tokio::io::{AsyncReadExt, AsyncWriteExt}; diff --git a/discovery/src/dns.rs b/discovery/src/dns.rs index 5329618a7..6ae7f6208 100644 --- a/discovery/src/dns.rs +++ b/discovery/src/dns.rs @@ -166,7 +166,7 @@ pub async fn start_dns_resolver_refresher() { use std::task::Poll; let noop = noop_waker::noop_waker(); let mut ctx = std::task::Context::from_waker(&noop); - use std::time::{Duration, Instant}; + use ds::time::{Duration, Instant}; const CYCLE: Duration = Duration::from_secs(79); let mut tick = tokio::time::interval(Duration::from_secs(1)); let mut last = Instant::now(); // 上一次刷新的时间 diff --git a/discovery/src/update.rs b/discovery/src/update.rs index e93f61e60..5b91fa479 100644 --- a/discovery/src/update.rs +++ b/discovery/src/update.rs @@ -1,7 +1,7 @@ // 定期更新discovery. use super::{Discover, ServiceId, TopologyWrite}; use ds::chan::Receiver; -use std::time::{Duration, Instant}; +use ds::time::{Duration, Instant}; use tokio::time::interval; use crate::cache::DiscoveryCache; diff --git a/ds/src/mem/policy.rs b/ds/src/mem/policy.rs index 8ab5acf8f..0e5f081c2 100644 --- a/ds/src/mem/policy.rs +++ b/ds/src/mem/policy.rs @@ -1,5 +1,5 @@ const BUF_MIN: usize = 2 * 1024; -use std::time::Instant; +use crate::time::Instant; // 内存需要缩容时的策略 // 为了避免频繁的缩容，需要设置一个最小频繁，通常使用最小间隔时间 pub struct MemPolicy { @@ -108,8 +108,8 @@ impl Display for MemPolicy { #[cfg(debug_assertions)] mod trace { + use crate::time::Instant; use std::fmt::{self, Debug, Formatter}; - use std::time::Instant; pub(super) struct Trace { direction: &'static str, // 方向: true为tx, false为rx. 打日志用 id: usize, diff --git a/endpoint/src/cacheservice/config.rs b/endpoint/src/cacheservice/config.rs index 507017e0b..bcac2754b 100644 --- a/endpoint/src/cacheservice/config.rs +++ b/endpoint/src/cacheservice/config.rs @@ -1,6 +1,6 @@ use serde::{Deserialize, Serialize}; use sharding::hash; -//use std::time::Duration; +//use ds::time::Duration; #[derive(Serialize, Deserialize, Clone, Debug, Default, Hash)] pub struct Namespace { diff --git a/endpoint/src/cacheservice/topo.rs b/endpoint/src/cacheservice/topo.rs index de8d6a1a2..ca65aba74 100644 --- a/endpoint/src/cacheservice/topo.rs +++ b/endpoint/src/cacheservice/topo.rs @@ -3,7 +3,7 @@ use protocol::{Builder, Endpoint, Protocol, Request, Resource, Topology, TryNext use sharding::hash::Hasher; use sharding::Distance; use std::collections::HashMap; -use std::time::Duration; +use ds::time::Duration; use crate::TimeoutAdjust; use stream::Shards; diff --git a/endpoint/src/lib.rs b/endpoint/src/lib.rs index aed988578..2d4bdd7c8 100644 --- a/endpoint/src/lib.rs +++ b/endpoint/src/lib.rs @@ -24,7 +24,7 @@ trait TimeoutAdjust: Sized { } } -use std::time::Duration; +use ds::time::Duration; impl TimeoutAdjust for Duration { fn adjust(&mut self, ms: u32) { if ms > 0 { diff --git a/endpoint/src/phantomservice/config.rs b/endpoint/src/phantomservice/config.rs index ebf8423ff..fa1632ff6 100644 --- a/endpoint/src/phantomservice/config.rs +++ b/endpoint/src/phantomservice/config.rs @@ -1,4 +1,4 @@ -//use std::time::Duration; +//use ds::time::Duration; use serde::{Deserialize, Serialize}; diff --git a/endpoint/src/redisservice/config.rs b/endpoint/src/redisservice/config.rs index 111dfeaa2..ab549d57b 100644 --- a/endpoint/src/redisservice/config.rs +++ b/endpoint/src/redisservice/config.rs @@ -1,5 +1,5 @@ use std::collections::{HashMap, HashSet}; -//use std::time::Duration; +//use ds::time::Duration; use serde::{Deserialize, Serialize}; //use sharding::distribution::{DIST_ABS_MODULA, DIST_MODULA}; diff --git a/endpoint/src/redisservice/topo.rs b/endpoint/src/redisservice/topo.rs index cdc1e17d8..cab5d67da 100644 --- a/endpoint/src/redisservice/topo.rs +++ b/endpoint/src/redisservice/topo.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; -use std::time::Duration; +use ds::time::Duration; use discovery::TopologyWrite; use protocol::{Builder, Endpoint, Protocol, Request, Resource, Single, Topology}; diff --git a/metrics/src/macros.rs b/metrics/src/macros.rs index 36e7fdca1..735ab80ef 100644 --- a/metrics/src/macros.rs +++ b/metrics/src/macros.rs @@ -42,7 +42,7 @@ impl MetricData for T { crate::register_cache(id, self.int()); } } -use std::time::Duration; +use ds::time::Duration; impl MetricData for Duration { #[inline] fn incr_to(self, data: &ItemData) { diff --git a/metrics/src/register.rs b/metrics/src/register.rs index 8a94c8271..59a683da2 100644 --- a/metrics/src/register.rs +++ b/metrics/src/register.rs @@ -168,7 +168,7 @@ pub struct MetricRegister { impl MetricRegister { fn new(rx: Receiver<(Arc, i64)>, metrics: CowWriteHandle) -> Self { - let mut tick = interval(std::time::Duration::from_secs(3)); + let mut tick = interval(ds::time::Duration::from_secs(3)); tick.set_missed_tick_behavior(MissedTickBehavior::Skip); Self { rx, diff --git a/metrics/src/sender.rs b/metrics/src/sender.rs index 109c83fb2..8ba291674 100644 --- a/metrics/src/sender.rs +++ b/metrics/src/sender.rs @@ -2,7 +2,7 @@ //use std::future::Future; //use std::pin::Pin; //use std::task::{Context, Poll}; -//use std::time::{Duration, Instant}; +//use ds::time::{Duration, Instant}; // //use std::task::ready; //use tokio::time::{interval, Interval, MissedTickBehavior}; diff --git a/metrics/src/types/host.rs b/metrics/src/types/host.rs index a009a4560..2b425d944 100644 --- a/metrics/src/types/host.rs +++ b/metrics/src/types/host.rs @@ -2,7 +2,7 @@ use psutil::process::Process; use crate::BASE_PATH; use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; -use std::time::Instant; +use ds::time::Instant; static CPU_PERCENT: AtomicUsize = AtomicUsize::new(0); static MEMORY: AtomicI64 = AtomicI64::new(0); diff --git a/metrics/src/types/rtt.rs b/metrics/src/types/rtt.rs index 203e40418..45169fb12 100644 --- a/metrics/src/types/rtt.rs +++ b/metrics/src/types/rtt.rs @@ -1,4 +1,4 @@ -use std::time::Duration; +use ds::time::Duration; use crate::{Id, ItemWriter, NumberInner}; pub const MAX: Duration = Duration::from_millis(30); diff --git a/protocol/src/callback.rs b/protocol/src/callback.rs index 3e6ab9d9d..7bbc9a4de 100644 --- a/protocol/src/callback.rs +++ b/protocol/src/callback.rs @@ -1,6 +1,6 @@ use std::mem::MaybeUninit; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::time::Instant; +use ds::time::Instant; use ds::AtomicWaker; diff --git a/protocol/src/error.rs b/protocol/src/error.rs index 1449ce1c0..e6fce7be8 100644 --- a/protocol/src/error.rs +++ b/protocol/src/error.rs @@ -1,4 +1,4 @@ -use std::time::Duration; +use ds::time::Duration; #[derive(Debug)] pub enum Error { diff --git a/protocol/src/lib.rs b/protocol/src/lib.rs index c4756e8ef..33225df75 100644 --- a/protocol/src/lib.rs +++ b/protocol/src/lib.rs @@ -47,7 +47,7 @@ impl Resource { } } -use std::time::Duration; +use ds::time::Duration; pub trait Builder { fn build(addr: &str, parser: P, rsrc: Resource, service: &str, timeout: Duration) -> E where diff --git a/protocol/src/req.rs b/protocol/src/req.rs index 2fe9cf90e..329dead4c 100644 --- a/protocol/src/req.rs +++ b/protocol/src/req.rs @@ -1,5 +1,5 @@ use std::fmt::{Debug, Display}; -use std::time::Instant; +use ds::time::Instant; use ds::RingSlice; diff --git a/protocol/src/request.rs b/protocol/src/request.rs index cf98d239c..2d45a76dd 100644 --- a/protocol/src/request.rs +++ b/protocol/src/request.rs @@ -7,7 +7,7 @@ pub struct Request { impl crate::Request for Request { #[inline] - fn start_at(&self) -> std::time::Instant { + fn start_at(&self) -> ds::time::Instant { self.ctx().start_at() } diff --git a/rt/src/entry.rs b/rt/src/entry.rs index dc66eb361..0f1b94a60 100644 --- a/rt/src/entry.rs +++ b/rt/src/entry.rs @@ -2,7 +2,7 @@ use std::fmt::Debug; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; -use std::time::{Duration, Instant}; +use ds::time::{Duration, Instant}; use metrics::{Metric, Path}; use std::task::ready; diff --git a/stream/src/builder.rs b/stream/src/builder.rs index 9c2f135a2..18d9cbccc 100644 --- a/stream/src/builder.rs +++ b/stream/src/builder.rs @@ -17,7 +17,7 @@ pub struct BackendBuilder { _marker: std::marker::PhantomData<(P, R)>, } -use std::time::Duration; +use ds::time::Duration; impl protocol::Builder>> for BackendBuilder { fn build( addr: &str, diff --git a/stream/src/checker.rs b/stream/src/checker.rs index 0a96beaec..2a23533da 100644 --- a/stream/src/checker.rs +++ b/stream/src/checker.rs @@ -1,5 +1,5 @@ use std::sync::{atomic::AtomicBool, Arc}; -use std::time::Duration; +use ds::time::Duration; use tokio::net::TcpStream; use tokio::time::timeout; diff --git a/stream/src/pipeline.rs b/stream/src/pipeline.rs index 64e4e18dc..e280e139f 100644 --- a/stream/src/pipeline.rs +++ b/stream/src/pipeline.rs @@ -3,7 +3,7 @@ use std::future::Future; use std::pin::Pin; use std::sync::atomic::{AtomicUsize, Ordering::*}; use std::task::{ready, Context, Poll}; -use std::time::{Duration, Instant}; +use ds::time::{Duration, Instant}; use tokio::io::{AsyncRead, AsyncWrite}; diff --git a/stream/src/reconn.rs b/stream/src/reconn.rs index 988e8d962..b99218d91 100644 --- a/stream/src/reconn.rs +++ b/stream/src/reconn.rs @@ -1,7 +1,7 @@ use metrics::{Metric, Path}; use std::sync::atomic::{AtomicBool, Ordering::Acquire}; use std::sync::Arc; -use std::time::Duration; +use ds::time::Duration; pub(crate) struct ReconnPolicy { single: Arc, conns: usize, diff --git a/tests/src/asserts.rs b/tests/src/asserts.rs new file mode 100644 index 000000000..13bf00335 --- /dev/null +++ b/tests/src/asserts.rs @@ -0,0 +1,4 @@ +#[test] +fn test_assert() { + ds::assert!(1 == 1, "assert true"); +} diff --git a/tests/src/mem.rs b/tests/src/mem.rs index feb973664..5d01a9a6f 100644 --- a/tests/src/mem.rs +++ b/tests/src/mem.rs @@ -63,7 +63,7 @@ fn ring_buffer() { rrb.advance_write(1024); // 等待10ms。（默认是4ms） - std::thread::sleep(std::time::Duration::from_millis(10)); + std::thread::sleep(ds::time::Duration::from_millis(10)); let buf = rrb.as_mut_bytes(); assert_eq!(buf.len(), 1024); rrb.advance_write(1024); diff --git a/tests/src/ring_buffer.rs b/tests/src/ring_buffer.rs index 435d7c155..7260945b3 100644 --- a/tests/src/ring_buffer.rs +++ b/tests/src/ring_buffer.rs @@ -1,7 +1,7 @@ #[cfg(test)] mod tests { use ds::{RingBuffer, RingSlice}; - use std::time::{Duration, Instant}; + use ds::time::{Duration, Instant}; fn rnd_bytes(size: usize) -> Vec { let data: Vec = (0..size).map(|_| rand::random::()).collect(); @@ -64,7 +64,7 @@ mod tests { assert_eq!(buf.len(), 0); // 等待10ms。（默认是4ms） - std::thread::sleep(std::time::Duration::from_millis(10)); + std::thread::sleep(ds::time::Duration::from_millis(10)); let buf = rrb.as_mut_bytes(); assert_eq!(buf.len(), 1024); rrb.advance_write(1024); @@ -76,7 +76,7 @@ mod tests { let ins = Instant::now(); loop { rrb.advance_write(0); - std::thread::sleep(std::time::Duration::from_millis(3)); + std::thread::sleep(ds::time::Duration::from_millis(3)); if ins.elapsed() >= Duration::from_secs(70) { break; }