Use single-threaded hashmap instead of dashmap
Address performance regressions
This commit is contained in:
1 parent
f4160d2387
commit
d7806043e1
3 files changed
+9
-29
No files matched your search
Generated
-21
@@ -442,7 +442,6 @@ dependencies = [
|
|||||||
"criterion",
|
"criterion",
|
||||||
"dagrs",
|
"dagrs",
|
||||||
"dagx-macros",
|
"dagx-macros",
|
||||||
"dashmap",
|
|
||||||
"futures",
|
"futures",
|
||||||
"futures-executor",
|
"futures-executor",
|
||||||
"parking_lot",
|
"parking_lot",
|
||||||
@@ -462,20 +461,6 @@ dependencies = [
|
|||||||
"syn 2.0.106",
|
"syn 2.0.106",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "dashmap"
|
|
||||||
version = "6.1.0"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf"
|
|
||||||
dependencies = [
|
|
||||||
"cfg-if",
|
|
||||||
"crossbeam-utils",
|
|
||||||
"hashbrown",
|
|
||||||
"lock_api",
|
|
||||||
"once_cell",
|
|
||||||
"parking_lot_core",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "either"
|
name = "either"
|
||||||
version = "1.15.0"
|
version = "1.15.0"
|
||||||
@@ -656,12 +641,6 @@ dependencies = [
|
|||||||
"crunchy",
|
"crunchy",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "hashbrown"
|
|
||||||
version = "0.14.5"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "hermit-abi"
|
name = "hermit-abi"
|
||||||
version = "0.5.2"
|
version = "0.5.2"
|
||||||
|
|||||||
@@ -32,7 +32,6 @@ futures = "0.3"
|
|||||||
parking_lot = "0.12"
|
parking_lot = "0.12"
|
||||||
dagx-macros = { version = "0.3.1", path = "./dagx-macros" }
|
dagx-macros = { version = "0.3.1", path = "./dagx-macros" }
|
||||||
tracing = { version = "0.1", optional = true }
|
tracing = { version = "0.1", optional = true }
|
||||||
dashmap = "6.1.0"
|
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
tokio = { version = "1", features = [
|
tokio = { version = "1", features = [
|
||||||
|
|||||||
+9
-7
@@ -13,7 +13,6 @@ use std::sync::{
|
|||||||
Arc,
|
Arc,
|
||||||
};
|
};
|
||||||
|
|
||||||
use dashmap::DashMap;
|
|
||||||
use futures::stream::FuturesUnordered;
|
use futures::stream::FuturesUnordered;
|
||||||
use futures::{FutureExt, StreamExt, TryFutureExt};
|
use futures::{FutureExt, StreamExt, TryFutureExt};
|
||||||
use parking_lot::Mutex;
|
use parking_lot::Mutex;
|
||||||
@@ -314,7 +313,7 @@ impl DagRunner {
|
|||||||
|
|
||||||
let edges = self.edges.lock().clone();
|
let edges = self.edges.lock().clone();
|
||||||
|
|
||||||
let outputs: DashMap<NodeId, Arc<dyn Any + Send + Sync>> = DashMap::new();
|
let mut outputs: HashMap<NodeId, Arc<dyn Any + Send + Sync>> = HashMap::new();
|
||||||
let mut first_error = None;
|
let mut first_error = None;
|
||||||
|
|
||||||
for layer in layers {
|
for layer in layers {
|
||||||
@@ -414,7 +413,6 @@ impl DagRunner {
|
|||||||
} else {
|
} else {
|
||||||
// Slow path: Multiple tasks require spawning and coordination
|
// Slow path: Multiple tasks require spawning and coordination
|
||||||
// Spawn each task in this layer
|
// Spawn each task in this layer
|
||||||
let outputs = &outputs;
|
|
||||||
let mut futures: FuturesUnordered<_> = FuturesUnordered::new();
|
let mut futures: FuturesUnordered<_> = FuturesUnordered::new();
|
||||||
layer
|
layer
|
||||||
.into_iter()
|
.into_iter()
|
||||||
@@ -437,8 +435,7 @@ impl DagRunner {
|
|||||||
// Spawn the task using the provided spawner
|
// Spawn the task using the provided spawner
|
||||||
let inner_future = async move {
|
let inner_future = async move {
|
||||||
let result = node.execute_with_deps(dependencies).await?;
|
let result = node.execute_with_deps(dependencies).await?;
|
||||||
outputs.insert(node_id, result);
|
Ok((node_id, result))
|
||||||
Ok(())
|
|
||||||
};
|
};
|
||||||
|
|
||||||
Some(
|
Some(
|
||||||
@@ -477,8 +474,13 @@ impl DagRunner {
|
|||||||
.for_each(|fut| futures.push(fut));
|
.for_each(|fut| futures.push(fut));
|
||||||
|
|
||||||
while let Some(out) = futures.next().await {
|
while let Some(out) = futures.next().await {
|
||||||
if let Err(e) = out {
|
match out {
|
||||||
first_error.get_or_insert(e);
|
Ok(output) => {
|
||||||
|
outputs.insert(output.0, output.1);
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
first_error.get_or_insert(e);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user