gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-25 13:36:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: migrate to reed-solomon-simd only implementation

Browse Source 
- Remove reed-solomon-erasure dependency and all related code
- Simplify ReedSolomonEncoder from enum to struct with SIMD-only implementation
- Eliminate all conditional compilation (#[cfg(feature = ...)])
- Add instance caching with RwLock-based encoder/decoder reuse
- Implement reset mechanism to avoid unnecessary allocations
- Ensure thread safety with proper cache management
- Update documentation and benchmark scripts for SIMD-only approach
- Apply code formatting across all files

Breaking Changes:
- Removes support for reed-solomon-erasure feature flag
- API remains compatible but implementation is now SIMD-only

Performance Impact:
- Improved encoding/decoding performance through SIMD optimization
- Reduced memory allocations via instance caching
- Enhanced thread safety and concurrency support
This commit is contained in:
weisd
2025-06-23 10:00:17 +08:00
parent 1722780560
commit 4559baaeeb
57 changed files with 404 additions and 728 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/notify/src/arn.rs
+5 -12
View File
@@ -41,7 +41,7 @@ impl TargetID {
ARN {
target_id: self.clone(),
region: region.to_string(),
service: DEFAULT_ARN_SERVICE.to_string(), // Default Service
service: DEFAULT_ARN_SERVICE.to_string(), // Default Service
partition: DEFAULT_ARN_PARTITION.to_string(), // Default partition
}
}
@@ -112,7 +112,7 @@ impl ARN {
ARN {
target_id,
region,
service: DEFAULT_ARN_SERVICE.to_string(), // Default is sqs
service: DEFAULT_ARN_SERVICE.to_string(), // Default is sqs
partition: DEFAULT_ARN_PARTITION.to_string(), // Default is rustfs partition
}
}
@@ -121,16 +121,10 @@ impl ARN {
/// Returns the ARN string in the format "{ARN_PREFIX}:{region}:{target_id}"
#[allow(clippy::inherent_to_string)]
pub fn to_arn_string(&self) -> String {
if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty()
{
if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty() {
return String::new();
}
format!(
"{}:{}:{}",
ARN_PREFIX,
self.region,
self.target_id.to_id_string()
)
format!("{}:{}:{}", ARN_PREFIX, self.region, self.target_id.to_id_string())
}
/// Parsing ARN from string
@@ -162,8 +156,7 @@ impl ARN {
impl fmt::Display for ARN {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty()
{
if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty() {
// Returns an empty string if all parts are empty
return Ok(());
}
crates/notify/src/factory.rs
+1 -1
View File
@@ -1,7 +1,7 @@
use crate::store::DEFAULT_LIMIT;
use crate::{
error::TargetError,
target::{mqtt::MQTTArgs, webhook::WebhookArgs, Target},
target::{Target, mqtt::MQTTArgs, webhook::WebhookArgs},
};
use async_trait::async_trait;
use ecstore::config::{ENABLE_KEY, ENABLE_ON, KVS};
crates/notify/src/integration.rs
+4 -4
View File
@@ -1,15 +1,15 @@
use crate::arn::TargetID;
use crate::store::{Key, Store};
use crate::{
error::NotificationError, notifier::EventNotifier, registry::TargetRegistry, rules::BucketNotificationConfig, stream, Event,
StoreError, Target,
Event, StoreError, Target, error::NotificationError, notifier::EventNotifier, registry::TargetRegistry,
rules::BucketNotificationConfig, stream,
};
use ecstore::config::{Config, KVS};
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, RwLock, Semaphore};
use tokio::sync::{RwLock, Semaphore, mpsc};
use tracing::{debug, error, info, warn};
/// Notify the system of monitoring indicators
crates/notify/src/lib.rs
+1 -1
View File
@@ -26,7 +26,7 @@ pub use rules::BucketNotificationConfig;
use std::io::IsTerminal;
pub use target::Target;
use tracing_subscriber::{fmt, prelude::*, util::SubscriberInitExt, EnvFilter};
use tracing_subscriber::{EnvFilter, fmt, prelude::*, util::SubscriberInitExt};
/// Initialize the tracing log system
///
crates/notify/src/notifier.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use crate::arn::TargetID;
use crate::{error::NotificationError, event::Event, rules::RulesMap, target::Target, EventName};
use crate::{EventName, error::NotificationError, event::Event, rules::RulesMap, target::Target};
use std::{collections::HashMap, sync::Arc};
use tokio::sync::RwLock;
use tracing::{debug, error, info, instrument, warn};
crates/notify/src/rules/config.rs
+2 -2
View File
@@ -1,11 +1,11 @@
use super::rules_map::RulesMap;
// Keep for existing structure if any, or remove if not used
use super::xml_config::ParseConfigError as BucketNotificationConfigError;
use crate::EventName;
use crate::arn::TargetID;
use crate::rules::NotificationConfiguration;
use crate::rules::pattern_rules;
use crate::rules::target_id_set;
use crate::rules::NotificationConfiguration;
use crate::EventName;
use std::collections::HashMap;
use std::io::Read;
// Assuming this is the XML config structure
crates/notify/src/rules/pattern.rs
+1 -4
View File
@@ -78,10 +78,7 @@ mod tests {
assert_eq!(new_pattern(Some(""), Some("b")), "*b");
assert_eq!(new_pattern(None, None), "");
assert_eq!(new_pattern(Some("prefix"), Some("suffix")), "prefix*suffix");
assert_eq!(
new_pattern(Some("prefix/"), Some("/suffix")),
"prefix/*suffix"
); // prefix/* + */suffix -> prefix/**/suffix -> prefix/*/suffix
assert_eq!(new_pattern(Some("prefix/"), Some("/suffix")), "prefix/*suffix"); // prefix/* + */suffix -> prefix/**/suffix -> prefix/*/suffix
}
#[test]
crates/notify/src/rules/pattern_rules.rs
+3 -8
View File
@@ -23,9 +23,7 @@ impl PatternRules {
/// Checks if there are any rules that match the given object name.
pub fn match_simple(&self, object_name: &str) -> bool {
self.rules
.keys()
.any(|p| pattern::match_simple(p, object_name))
self.rules.keys().any(|p| pattern::match_simple(p, object_name))
}
/// Returns all TargetIDs that match the object name.
@@ -61,8 +59,7 @@ impl PatternRules {
for (pattern, self_targets) in &self.rules {
match other.rules.get(pattern) {
Some(other_targets) => {
let diff_targets: TargetIdSet =
self_targets.difference(other_targets).cloned().collect();
let diff_targets: TargetIdSet = self_targets.difference(other_targets).cloned().collect();
if !diff_targets.is_empty() {
result_rules.insert(pattern.clone(), diff_targets);
}
@@ -73,8 +70,6 @@ impl PatternRules {
}
}
}
PatternRules {
rules: result_rules,
}
PatternRules { rules: result_rules }
}
}
crates/notify/src/rules/xml_config.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use super::pattern;
use crate::arn::{ArnError, TargetIDError, ARN};
use crate::arn::{ARN, ArnError, TargetIDError};
use crate::event::EventName;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
crates/notify/src/store.rs
+35 -50
View File
@@ -1,5 +1,5 @@
use crate::error::StoreError;
use serde::{de::DeserializeOwned, Serialize};
use serde::{Serialize, de::DeserializeOwned};
use snap::raw::{Decoder, Encoder};
use std::sync::{Arc, RwLock};
use std::{
@@ -195,11 +195,7 @@ impl<T: Serialize + DeserializeOwned + Send + Sync> QueueStore<T> {
/// Reads a file for the given key
fn read_file(&self, key: &Key) -> Result<Vec<u8>, StoreError> {
let path = self.file_path(key);
debug!(
"Reading file for key: {},path: {}",
key.to_string(),
path.display()
);
debug!("Reading file for key: {},path: {}", key.to_string(), path.display());
let data = std::fs::read(&path).map_err(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
StoreError::NotFound
@@ -240,13 +236,11 @@ impl<T: Serialize + DeserializeOwned + Send + Sync> QueueStore<T> {
};
std::fs::write(&path, &data).map_err(StoreError::Io)?;
let modified = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as i64;
let mut entries = self.entries.write().map_err(|_| {
StoreError::Internal("Failed to acquire write lock on entries".to_string())
})?;
let modified = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
let mut entries = self
.entries
.write()
.map_err(|_| StoreError::Internal("Failed to acquire write lock on entries".to_string()))?;
entries.insert(key.to_string(), modified);
debug!("Wrote event to store: {}", key.to_string());
Ok(())
@@ -265,18 +259,16 @@ where
let entries = std::fs::read_dir(&self.directory).map_err(StoreError::Io)?;
// Get the write lock to update the internal state
let mut entries_map = self.entries.write().map_err(|_| {
StoreError::Internal("Failed to acquire write lock on entries".to_string())
})?;
let mut entries_map = self
.entries
.write()
.map_err(|_| StoreError::Internal("Failed to acquire write lock on entries".to_string()))?;
for entry in entries {
let entry = entry.map_err(StoreError::Io)?;
let metadata = entry.metadata().map_err(StoreError::Io)?;
if metadata.is_file() {
let modified = metadata.modified().map_err(StoreError::Io)?;
let unix_nano = modified
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as i64;
let unix_nano = modified.duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
let file_name = entry.file_name().to_string_lossy().to_string();
entries_map.insert(file_name, unix_nano);
@@ -290,9 +282,10 @@ where
fn put(&self, item: T) -> Result<Self::Key, Self::Error> {
// Check storage limits
{
let entries = self.entries.read().map_err(|_| {
StoreError::Internal("Failed to acquire read lock on entries".to_string())
})?;
let entries = self
.entries
.read()
.map_err(|_| StoreError::Internal("Failed to acquire read lock on entries".to_string()))?;
if entries.len() as u64 >= self.entry_limit {
return Err(StoreError::LimitExceeded);
@@ -307,8 +300,7 @@ where
compress: true,
};
let data =
serde_json::to_vec(&item).map_err(|e| StoreError::Serialization(e.to_string()))?;
let data = serde_json::to_vec(&item).map_err(|e| StoreError::Serialization(e.to_string()))?;
self.write_file(&key, &data)?;
Ok(key)
@@ -317,9 +309,10 @@ where
fn put_multiple(&self, items: Vec<T>) -> Result<Self::Key, Self::Error> {
// Check storage limits
{
let entries = self.entries.read().map_err(|_| {
StoreError::Internal("Failed to acquire read lock on entries".to_string())
})?;
let entries = self
.entries
.read()
.map_err(|_| StoreError::Internal("Failed to acquire read lock on entries".to_string()))?;
if entries.len() as u64 >= self.entry_limit {
return Err(StoreError::LimitExceeded);
@@ -327,9 +320,7 @@ where
}
if items.is_empty() {
// Or return an error, or a special key?
return Err(StoreError::Internal(
"Cannot put_multiple with empty items list".to_string(),
));
return Err(StoreError::Internal("Cannot put_multiple with empty items list".to_string()));
}
let uuid = Uuid::new_v4();
let key = Key {
@@ -348,8 +339,7 @@ where
for item in items {
// If items are Vec<Event>, and Event is large, this could be inefficient.
// The current get_multiple deserializes one by one.
let item_data =
serde_json::to_vec(&item).map_err(|e| StoreError::Serialization(e.to_string()))?;
let item_data = serde_json::to_vec(&item).map_err(|e| StoreError::Serialization(e.to_string()))?;
buffer.extend_from_slice(&item_data);
// If using JSON array: buffer = serde_json::to_vec(&items)?
}
@@ -374,9 +364,7 @@ where
debug!("Reading items from store for key: {}", key.to_string());
let data = self.read_file(key)?;
if data.is_empty() {
return Err(StoreError::Deserialization(
"Cannot deserialize empty data".to_string(),
));
return Err(StoreError::Deserialization("Cannot deserialize empty data".to_string()));
}
let mut items = Vec::with_capacity(key.item_count);
@@ -395,10 +383,7 @@ where
match deserializer.next() {
Some(Ok(item)) => items.push(item),
Some(Err(e)) => {
return Err(StoreError::Deserialization(format!(
"Failed to deserialize item in batch: {}",
e
)));
return Err(StoreError::Deserialization(format!("Failed to deserialize item in batch: {}", e)));
}
None => {
// Reached end of stream sooner than item_count
@@ -435,7 +420,10 @@ where
std::fs::remove_file(&path).map_err(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
// If file not found, still try to remove from entries map in case of inconsistency
warn!("File not found for key {} during del, but proceeding to remove from entries map.", key.to_string());
warn!(
"File not found for key {} during del, but proceeding to remove from entries map.",
key.to_string()
);
StoreError::NotFound
} else {
StoreError::Io(e)
@@ -443,17 +431,15 @@ where
})?;
// Get the write lock to update the internal state
let mut entries = self.entries.write().map_err(|_| {
StoreError::Internal("Failed to acquire write lock on entries".to_string())
})?;
let mut entries = self
.entries
.write()
.map_err(|_| StoreError::Internal("Failed to acquire write lock on entries".to_string()))?;
if entries.remove(&key.to_string()).is_none() {
// Key was not in the map, could be an inconsistency or already deleted.
// This is not necessarily an error if the file deletion succeeded or was NotFound.
debug!(
"Key {} not found in entries map during del, might have been already removed.",
key
);
debug!("Key {} not found in entries map during del, might have been already removed.", key);
}
debug!("Deleted event from store: {}", key.to_string());
Ok(())
@@ -492,7 +478,6 @@ where
}
fn boxed_clone(&self) -> Box<dyn Store<T, Error = Self::Error, Key = Self::Key> + Send + Sync> {
Box::new(self.clone())
as Box<dyn Store<T, Error = Self::Error, Key = Self::Key> + Send + Sync>
Box::new(self.clone()) as Box<dyn Store<T, Error = Self::Error, Key = Self::Key> + Send + Sync>
}
}
crates/notify/src/stream.rs
+4 -4
View File
@@ -1,13 +1,13 @@
use crate::{
error::TargetError, integration::NotificationMetrics,
Event, StoreError,
error::TargetError,
integration::NotificationMetrics,
store::{Key, Store},
target::Target,
Event,
StoreError,
};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, Semaphore};
use tokio::sync::{Semaphore, mpsc};
use tokio::time::sleep;
use tracing::{debug, error, info, warn};
crates/notify/src/target/mqtt.rs
+25 -66
View File
@@ -1,22 +1,22 @@
use crate::store::{Key, STORE_EXTENSION};
use crate::target::ChannelTargetType;
use crate::{
arn::TargetID, error::TargetError,
StoreError, Target,
arn::TargetID,
error::TargetError,
event::{Event, EventLog},
store::Store,
StoreError,
Target,
};
use async_trait::async_trait;
use rumqttc::{mqttbytes::Error as MqttBytesError, ConnectionError};
use rumqttc::{AsyncClient, EventLoop, MqttOptions, Outgoing, Packet, QoS};
use rumqttc::{ConnectionError, mqttbytes::Error as MqttBytesError};
use std::sync::Arc;
use std::{
path::PathBuf,
sync::atomic::{AtomicBool, Ordering},
time::Duration,
};
use tokio::sync::{mpsc, Mutex, OnceCell};
use tokio::sync::{Mutex, OnceCell, mpsc};
use tracing::{debug, error, info, instrument, trace, warn};
use url::Url;
use urlencoding;
@@ -58,24 +58,19 @@ impl MQTTArgs {
match self.broker.scheme() {
"ws" | "wss" | "tcp" | "ssl" | "tls" | "tcps" | "mqtt" | "mqtts" => {}
_ => {
return Err(TargetError::Configuration(
"unknown protocol in broker address".to_string(),
));
return Err(TargetError::Configuration("unknown protocol in broker address".to_string()));
}
}
if !self.queue_dir.is_empty() {
let path = std::path::Path::new(&self.queue_dir);
if !path.is_absolute() {
return Err(TargetError::Configuration(
"mqtt queueDir path should be absolute".to_string(),
));
return Err(TargetError::Configuration("mqtt queueDir path should be absolute".to_string()));
}
if self.qos == QoS::AtMostOnce {
return Err(TargetError::Configuration(
"QoS should be AtLeastOnce (1) or ExactlyOnce (2) if queueDir is set"
.to_string(),
"QoS should be AtLeastOnce (1) or ExactlyOnce (2) if queueDir is set".to_string(),
));
}
}
@@ -107,21 +102,12 @@ impl MQTTTarget {
let target_id = TargetID::new(id.clone(), ChannelTargetType::Mqtt.as_str().to_string());
let queue_store = if !args.queue_dir.is_empty() {
let base_path = PathBuf::from(&args.queue_dir);
let unique_dir_name = format!(
"rustfs-{}-{}-{}",
ChannelTargetType::Mqtt.as_str(),
target_id.name,
target_id.id
)
.replace(":", "_");
let unique_dir_name =
format!("rustfs-{}-{}-{}", ChannelTargetType::Mqtt.as_str(), target_id.name, target_id.id).replace(":", "_");
// Ensure the directory name is valid for filesystem
let specific_queue_path = base_path.join(unique_dir_name);
debug!(target_id = %target_id, path = %specific_queue_path.display(), "Initializing queue store for MQTT target");
let store = crate::store::QueueStore::<Event>::new(
specific_queue_path,
args.queue_limit,
STORE_EXTENSION,
);
let store = crate::store::QueueStore::<Event>::new(specific_queue_path, args.queue_limit, STORE_EXTENSION);
if let Err(e) = store.open() {
error!(
target_id = %target_id,
@@ -130,10 +116,7 @@ impl MQTTTarget {
);
return Err(TargetError::Storage(format!("{}", e)));
}
Some(Box::new(store)
as Box<
dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync,
>)
Some(Box::new(store) as Box<dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync>)
} else {
None
};
@@ -175,18 +158,13 @@ impl MQTTTarget {
debug!(target_id = %target_id_clone, "Initializing MQTT background task.");
let host = args_clone.broker.host_str().unwrap_or("localhost");
let port = args_clone.broker.port().unwrap_or(1883);
let mut mqtt_options = MqttOptions::new(
format!("rustfs_notify_{}", uuid::Uuid::new_v4()),
host,
port,
);
let mut mqtt_options = MqttOptions::new(format!("rustfs_notify_{}", uuid::Uuid::new_v4()), host, port);
mqtt_options
.set_keep_alive(args_clone.keep_alive)
.set_max_packet_size(100 * 1024 * 1024, 100 * 1024 * 1024); // 100MB
if !args_clone.username.is_empty() {
mqtt_options
.set_credentials(args_clone.username.clone(), args_clone.password.clone());
mqtt_options.set_credentials(args_clone.username.clone(), args_clone.password.clone());
}
let (new_client, eventloop) = AsyncClient::new(mqtt_options, 10);
@@ -206,12 +184,8 @@ impl MQTTTarget {
*client_arc.lock().await = Some(new_client.clone());
info!(target_id = %target_id_clone, "Spawning MQTT event loop task.");
let task_handle = tokio::spawn(run_mqtt_event_loop(
eventloop,
connected_arc.clone(),
target_id_clone.clone(),
cancel_rx,
));
let task_handle =
tokio::spawn(run_mqtt_event_loop(eventloop, connected_arc.clone(), target_id_clone.clone(), cancel_rx));
Ok(task_handle)
})
.await
@@ -266,17 +240,13 @@ impl MQTTTarget {
records: vec![event.clone()],
};
let data = serde_json::to_vec(&log)
.map_err(|e| TargetError::Serialization(format!("Failed to serialize event: {}", e)))?;
let data =
serde_json::to_vec(&log).map_err(|e| TargetError::Serialization(format!("Failed to serialize event: {}", e)))?;
// Vec<u8> Convert to String, only for printing logs
let data_string = String::from_utf8(data.clone()).map_err(|e| {
TargetError::Encoding(format!("Failed to convert event data to UTF-8: {}", e))
})?;
debug!(
"Sending event to mqtt target: {}, event log: {}",
self.id, data_string
);
let data_string = String::from_utf8(data.clone())
.map_err(|e| TargetError::Encoding(format!("Failed to convert event data to UTF-8: {}", e)))?;
debug!("Sending event to mqtt target: {}, event log: {}", self.id, data_string);
client
.publish(&self.args.topic, self.args.qos, false, data)
@@ -474,9 +444,7 @@ impl Target for MQTTTarget {
if let Some(handle) = self.bg_task_manager.init_cell.get() {
if handle.is_finished() {
error!(target_id = %self.id, "MQTT background task has finished, possibly due to an error. Target is not active.");
return Err(TargetError::Network(
"MQTT background task terminated".to_string(),
));
return Err(TargetError::Network("MQTT background task terminated".to_string()));
}
}
debug!(target_id = %self.id, "MQTT client not yet initialized or task not running/connected.");
@@ -507,10 +475,7 @@ impl Target for MQTTTarget {
}
Err(e) => {
error!(target_id = %self.id, error = %e, "Failed to save event to store");
return Err(TargetError::Storage(format!(
"Failed to save event to store: {}",
e
)));
return Err(TargetError::Storage(format!("Failed to save event to store: {}", e)));
}
}
} else {
@@ -581,10 +546,7 @@ impl Target for MQTTTarget {
error = %e,
"Failed to get event from store"
);
return Err(TargetError::Storage(format!(
"Failed to get event from store: {}",
e
)));
return Err(TargetError::Storage(format!("Failed to get event from store: {}", e)));
}
};
@@ -608,10 +570,7 @@ impl Target for MQTTTarget {
}
Err(e) => {
error!(target_id = %self.id, error = %e, "Failed to delete event from store after send.");
return Err(TargetError::Storage(format!(
"Failed to delete event from store: {}",
e
)));
return Err(TargetError::Storage(format!("Failed to delete event from store: {}", e)));
}
}
crates/notify/src/target/webhook.rs
+4 -4
View File
@@ -1,19 +1,19 @@
use crate::store::STORE_EXTENSION;
use crate::target::ChannelTargetType;
use crate::{
arn::TargetID, error::TargetError,
StoreError, Target,
arn::TargetID,
error::TargetError,
event::{Event, EventLog},
store::{Key, Store},
StoreError,
Target,
};
use async_trait::async_trait;
use reqwest::{Client, StatusCode, Url};
use std::{
path::PathBuf,
sync::{
atomic::{AtomicBool, Ordering},
Arc,
atomic::{AtomicBool, Ordering},
},
time::Duration,
};