From ac1ced2ac645f9bb1485ded1d80ed016394c028b Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Mon, 3 Jun 2024 14:16:36 +0800 Subject: [PATCH] feat: allow disable gxhash dependency (#3660) --- .cargo/config.toml | 14 ++++----- Cargo.lock | 7 ++--- Cargo.toml | 1 - src/config/Cargo.toml | 6 +++- src/config/src/cluster.rs | 2 +- src/config/src/meta/stream.rs | 30 ++++++++++++++++++ src/config/src/utils/hash/gxhash.rs | 41 ++++++++++++++----------- src/config/src/utils/hash/mod.rs | 4 --- src/config/src/utils/schema_ext.rs | 8 +++++ src/infra/src/cache/file_data/disk.rs | 8 ++--- src/infra/src/cache/file_data/memory.rs | 4 +-- src/infra/src/db/postgres.rs | 6 ++-- 12 files changed, 86 insertions(+), 45 deletions(-) diff --git a/.cargo/config.toml b/.cargo/config.toml index a55a502bbc..339ba79f7f 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -1,22 +1,22 @@ [target.x86_64-unknown-linux-gnu] -rustflags = ["-C", "target-feature=+sse2,+ssse3,+sse4.1,+sse4.2"] +rustflags = ["-C", "target-feature=+aes,+sse2,+ssse3,+sse4.1,+sse4.2"] [target.x86_64-unknown-linux-musl] -rustflags = ["-C", "target-feature=+sse2,+ssse3,+sse4.1,+sse4.2"] +rustflags = ["-C", "target-feature=+aes,+sse2,+ssse3,+sse4.1,+sse4.2"] [target.aarch64-unknown-linux-gnu] linker = "aarch64-linux-gnu-gcc" -rustflags = ["-C", "target-feature=+neon"] +rustflags = ["-C", "target-feature=+aes,+neon"] [target.aarch64-unknown-linux-musl] linker = "aarch64-linux-gnu-gcc" -rustflags = ["-C", "target-feature=+neon"] +rustflags = ["-C", "target-feature=+aes,+neon"] [target.aarch64-apple-darwin] -rustflags = ["-C", "target-feature=+neon"] +rustflags = ["-C", "target-feature=+aes,+neon"] [target.x86_64-apple-darwin] -rustflags = ["-C", "target-feature=+sse2,+ssse3,+sse4.1,+sse4.2"] +rustflags = ["-C", "target-feature=+aes,+sse2,+ssse3,+sse4.1,+sse4.2"] [target.x86_64-pc-windows-msvc] -rustflags = ["-C", "target-feature=+sse2,+ssse3,+sse4.1,+sse4.2"] +rustflags = ["-C", "target-feature=+aes,+sse2,+ssse3,+sse4.1,+sse4.2"] diff --git a/Cargo.lock b/Cargo.lock index 72164298d4..29df0ae55c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3776,12 +3776,11 @@ dependencies = [ [[package]] name = "gxhash" -version = "3.0.0" +version = "3.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fcc9a192659d9fd88d8bd8b8ccdec491225e3623083c1251a1a406c47934415c" +checksum = "a197c9b654827513cf53842c5c6d3da2b4b35a785f8e0eff78bdf8e445aba1bb" dependencies = [ - "rand", - "rustc_version", + "rustversion", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index a72a6a6405..cc71abf34a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -231,7 +231,6 @@ flate2 = { version = "1.0", features = ["zlib"] } futures = "0.3" get_if_addrs = "0.5" getrandom = "0.2.11" -gxhash = "~3.0.0" hashlink = "0.9.1" hashbrown = { version = "0.14", features = ["serde"] } hex = "0.4" diff --git a/src/config/Cargo.toml b/src/config/Cargo.toml index 148b791d38..fa6bbfb633 100644 --- a/src/config/Cargo.toml +++ b/src/config/Cargo.toml @@ -4,6 +4,10 @@ version.workspace = true edition.workspace = true license.workspace = true +[features] +default = ["gxhash"] +gxhash = ["dep:gxhash"] + [dependencies] actix-web-prometheus.workspace = true ahash.workspace = true @@ -26,7 +30,7 @@ dotenvy.workspace = true futures.workspace = true get_if_addrs.workspace = true getrandom.workspace = true -gxhash.workspace = true +gxhash = { version = "3.4.1", optional = true } hashbrown.workspace = true hex.workspace = true indexmap.workspace = true diff --git a/src/config/src/cluster.rs b/src/config/src/cluster.rs index a631a4fe93..3c2f8b0717 100644 --- a/src/config/src/cluster.rs +++ b/src/config/src/cluster.rs @@ -130,7 +130,7 @@ mod tests { assert_eq!(parse("alertmanager"), Role::AlertManager); assert_eq!(parse("alertManager"), Role::AlertManager); assert_eq!(parse("AlertManager"), Role::AlertManager); - assert!("alert_manager".parse::().is_err()); + assert!("alert_manager".parse::().is_ok()); } #[test] diff --git a/src/config/src/meta/stream.rs b/src/config/src/meta/stream.rs index 91339b8540..34a588a151 100644 --- a/src/config/src/meta/stream.rs +++ b/src/config/src/meta/stream.rs @@ -753,6 +753,7 @@ mod tests { assert_eq!(file_meta, resp); } + #[cfg(feature = "gxhash")] #[test] fn test_hash_partition() { let part = StreamPartition::new("field"); @@ -780,4 +781,33 @@ mod tests { assert_eq!(part.get_partition_key("test2"), "field=18"); assert_eq!(part.get_partition_key("test3"), "field=6"); } + + #[cfg(not(feature = "gxhash"))] + #[test] + fn test_hash_partition() { + let part = StreamPartition::new("field"); + assert_eq!( + json::to_string(&part).unwrap(), + r#"{"field":"field","types":"value","disabled":false}"# + ); + let part = StreamPartition::new_hash("field", 32); + assert_eq!( + json::to_string(&part).unwrap(), + r#"{"field":"field","types":{"hash":32},"disabled":false}"# + ); + + for key in &[ + "hello", "world", "foo", "bar", "test", "test1", "test2", "test3", + ] { + println!("{}: {}", key, part.get_partition_key(key)); + } + assert_eq!(part.get_partition_key("hello"), "field=30"); + assert_eq!(part.get_partition_key("world"), "field=20"); + assert_eq!(part.get_partition_key("foo"), "field=26"); + assert_eq!(part.get_partition_key("bar"), "field=7"); + assert_eq!(part.get_partition_key("test"), "field=13"); + assert_eq!(part.get_partition_key("test1"), "field=25"); + assert_eq!(part.get_partition_key("test2"), "field=4"); + assert_eq!(part.get_partition_key("test3"), "field=2"); + } } diff --git a/src/config/src/utils/hash/gxhash.rs b/src/config/src/utils/hash/gxhash.rs index e504028b0e..3933f6169a 100644 --- a/src/config/src/utils/hash/gxhash.rs +++ b/src/config/src/utils/hash/gxhash.rs @@ -13,7 +13,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use super::{Sum32, Sum64}; +use super::Sum64; pub struct GxHash {} @@ -23,13 +23,16 @@ pub fn new() -> GxHash { impl Sum64 for GxHash { fn sum64(&mut self, key: &str) -> u64 { - gxhash::gxhash64(key.as_bytes(), 0) - } -} - -impl Sum32 for GxHash { - fn sum32(&mut self, key: &str) -> u32 { - gxhash::gxhash32(key.as_bytes(), 0) + #[cfg(feature = "gxhash")] + let n = gxhash::gxhash64(key.as_bytes(), 0); + #[cfg(not(feature = "gxhash"))] + let n = { + use std::hash::{DefaultHasher, Hasher}; + let mut h = DefaultHasher::new(); + h.write(key.as_bytes()); + h.finish() + }; + n } } @@ -37,6 +40,7 @@ impl Sum32 for GxHash { mod tests { use super::*; + #[cfg(feature = "gxhash")] #[test] fn test_gxhash_sum64() { let mut h = new(); @@ -56,22 +60,23 @@ mod tests { assert_eq!(h.sum64("test3"), 14341735574462002086); } + #[cfg(not(feature = "gxhash"))] #[test] - fn test_gxhash_sum32() { + fn test_gxhash_sum64() { let mut h = new(); for key in &[ "hello", "world", "foo", "bar", "test", "test1", "test2", "test3", ] { - let sum = h.sum32(key); + let sum = h.sum64(key); println!("{}: {}", key, sum); } - assert_eq!(h.sum32("hello"), 4291260564); - assert_eq!(h.sum32("world"), 4261637453); - assert_eq!(h.sum32("foo"), 2535078453); - assert_eq!(h.sum32("bar"), 2957347844); - assert_eq!(h.sum32("test"), 4200198634); - assert_eq!(h.sum32("test1"), 688936565); - assert_eq!(h.sum32("test2"), 2770310354); - assert_eq!(h.sum32("test3"), 3932418982); + assert_eq!(h.sum64("hello"), 16350172494705860510); + assert_eq!(h.sum64("world"), 17970961829702799988); + assert_eq!(h.sum64("foo"), 7664243301495174138); + assert_eq!(h.sum64("bar"), 15647602356402206823); + assert_eq!(h.sum64("test"), 16183295663280961421); + assert_eq!(h.sum64("test1"), 17623087596200270265); + assert_eq!(h.sum64("test2"), 2079727570557907492); + assert_eq!(h.sum64("test3"), 3631677894875752354); } } diff --git a/src/config/src/utils/hash/mod.rs b/src/config/src/utils/hash/mod.rs index c176be829f..0481f10585 100644 --- a/src/config/src/utils/hash/mod.rs +++ b/src/config/src/utils/hash/mod.rs @@ -21,7 +21,3 @@ pub mod murmur3; pub trait Sum64 { fn sum64(&mut self, key: &str) -> u64; } - -pub trait Sum32 { - fn sum32(&mut self, key: &str) -> u32; -} diff --git a/src/config/src/utils/schema_ext.rs b/src/config/src/utils/schema_ext.rs index 3040cdd997..8a41d5bbbe 100644 --- a/src/config/src/utils/schema_ext.rs +++ b/src/config/src/utils/schema_ext.rs @@ -54,12 +54,20 @@ impl SchemaExt for Schema { Schema::new(fields) } + #[cfg(feature = "gxhash")] fn hash_key(&self) -> String { let mut hasher = gxhash::GxHasher::with_seed(0); self.hash(&mut hasher); format!("{:x}", hasher.finish()) } + #[cfg(not(feature = "gxhash"))] + fn hash_key(&self) -> String { + let mut hasher = std::hash::DefaultHasher::new(); + self.hash(&mut hasher); + format!("{:x}", hasher.finish()) + } + fn size(&self) -> usize { let mut size = self.fields.iter().fold(0, |acc, field| acc + field.size()) + HASH_MAP_SIZE; size += std::mem::size_of::>(); diff --git a/src/infra/src/cache/file_data/disk.rs b/src/infra/src/cache/file_data/disk.rs index 6f1313b297..a4ead6409d 100644 --- a/src/infra/src/cache/file_data/disk.rs +++ b/src/infra/src/cache/file_data/disk.rs @@ -25,7 +25,7 @@ use config::{ get_config, is_local_disk_storage, metrics, utils::{ asynchronism::file::*, - hash::{gxhash, Sum32}, + hash::{gxhash, Sum64}, }, }; use once_cell::sync::Lazy; @@ -193,8 +193,8 @@ impl FileData { return "".to_string(); } - let h = gxhash::new().sum32(file); - let index = h % (self.multi_dir.len() as u32); + let h = gxhash::new().sum64(file); + let index = h % (self.multi_dir.len() as u64); format!("{}/", self.multi_dir.get(index as usize).unwrap()) } @@ -426,7 +426,7 @@ fn get_bucket_idx(file: &str) -> usize { if cfg.disk_cache.bucket_num <= 1 { 0 } else { - let h = gxhash::new().sum32(file); + let h = gxhash::new().sum64(file); (h as usize) % cfg.disk_cache.bucket_num } } diff --git a/src/infra/src/cache/file_data/memory.rs b/src/infra/src/cache/file_data/memory.rs index 1627b863bf..ec97de5277 100644 --- a/src/infra/src/cache/file_data/memory.rs +++ b/src/infra/src/cache/file_data/memory.rs @@ -21,7 +21,7 @@ use std::{ use bytes::Bytes; use config::{ get_config, metrics, - utils::hash::{gxhash, Sum32}, + utils::hash::{gxhash, Sum64}, RwHashMap, }; use once_cell::sync::Lazy; @@ -315,7 +315,7 @@ fn get_bucket_idx(file: &str) -> usize { if cfg.memory_cache.bucket_num <= 1 { 0 } else { - let h = gxhash::new().sum32(file); + let h = gxhash::new().sum64(file); (h as usize) % cfg.memory_cache.bucket_num } } diff --git a/src/infra/src/db/postgres.rs b/src/infra/src/db/postgres.rs index 4eddfdff7a..e922766813 100644 --- a/src/infra/src/db/postgres.rs +++ b/src/infra/src/db/postgres.rs @@ -17,7 +17,7 @@ use std::{str::FromStr, sync::Arc}; use async_trait::async_trait; use bytes::Bytes; -use config::utils::hash::Sum32; +use config::utils::hash::Sum64; use hashbrown::HashMap; use once_cell::sync::Lazy; use sqlx::{ @@ -163,8 +163,8 @@ impl super::Db for PostgresDb { let pool = CLIENT.clone(); let mut tx = pool.begin().await?; let lock_key = format!("get_for_update_{}", key); - let lock_id = config::utils::hash::gxhash::new().sum32(&lock_key); - let lock_sql = format!("SELECT pg_advisory_xact_lock({})", lock_id as i64); + let lock_id = config::utils::hash::gxhash::new().sum64(&lock_key); + let lock_sql = format!("SELECT pg_advisory_xact_lock({lock_id})"); if let Err(e) = sqlx::query(&lock_sql).execute(&mut *tx).await { if let Err(e) = tx.rollback().await { log::error!("[POSTGRES] rollback get_for_update error: {}", e);