logos-co
/
nomos-node
mirror of
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Finish `NodeId` type wrapper (#252) 

* add NodeId wrapper
This commit is contained in:
Al Liu 2023-07-11 23:16:49 +08:00 committed by GitHub
parent da2dba2e51
commit 2135676606
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
30 changed files with 303 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
consensus-engine/Cargo.toml
View File

@ -18,6 +18,7 @@ fraction = { version = "0.13" }
[features]
default = []
serde = ["dep:serde"]
simulation = []
[dev-dependencies]
proptest = "1.2.0"

60
consensus-engine/src/lib.rs
View File

@ -454,7 +454,7 @@ mod test {
#[test]
// Ensure that all states are initialized correctly with the genesis block.
fn from_genesis() {
let engine = init(vec![[0; 32]]);
let engine = init(vec![NodeId::new([0; 32])]);
assert_eq!(engine.current_view(), 0);
assert_eq!(engine.highest_voted_view, -1);
@ -468,7 +468,7 @@ mod test {
#[test]
// Ensure that all states are updated correctly after a block is received.
fn receive_block() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block.clone()).unwrap();
@ -480,7 +480,7 @@ mod test {
#[test]
// Ensure that receive_block() returns early if the same block ID has already been received.
fn receive_duplicate_block_id() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
let block1 = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block1.clone()).unwrap();
@ -496,7 +496,7 @@ mod test {
#[should_panic(expected = "out of order view not supported, missing parent block")]
// Ensure that receive_block() fails if the parent block has never been received.
fn receive_block_with_unknown_parent() {
let engine = init(vec![[0; 32]]);
let engine = init(vec![NodeId::new([0; 32])]);
let mut parent_block_id = engine.genesis_block().id;
parent_block_id[0] += 1; // generate an unknown parent block ID
let block = Block {
@ -506,7 +506,9 @@ mod test {
view: engine.current_view(),
id: parent_block_id,
}),
leader_proof: LeaderProof::LeaderId { leader_id: [0; 32] },
leader_proof: LeaderProof::LeaderId {
leader_id: NodeId::new([0; 32]),
},
};
let _ = engine.receive_block(block);
@ -515,7 +517,7 @@ mod test {
#[test]
// Ensure that receive_block() returns Err for unsafe blocks.
fn receive_unsafe_blocks() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block.clone()).unwrap();
@ -537,7 +539,7 @@ mod test {
#[test]
// Ensure that the grandparent of the current view can be committed
fn receive_block_and_commit() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
assert_eq!(engine.latest_committed_block(), engine.genesis_block());
let block1 = next_block(&engine, &engine.genesis_block());
@ -572,7 +574,7 @@ mod test {
// Ensure that the leader check in receive_block fails
// if the block is proposed by an unexpected leader.
fn receive_block_with_unexpected_leader() {
let mut engine = init(vec![[0; 32], [1; 32]]);
let mut engine = init(vec![NodeId::new([0; 32]), NodeId::new([1; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block.clone()).unwrap();
@ -581,7 +583,7 @@ mod test {
let mut block = next_block(&engine, &block);
block.leader_proof = LeaderProof::LeaderId {
leader_id: [0; 32], // unexpected leader
leader_id: NodeId::new([0; 32]), // unexpected leader
};
assert!(engine.receive_block(block).is_err());
}
@ -590,7 +592,7 @@ mod test {
// Ensure that the leader check in receive_block fails
// if block.view is not the expected view.
fn receive_block_with_unexpected_view() {
let mut engine = init(vec![[0; 32], [1; 32]]);
let mut engine = init(vec![NodeId::new([0; 32]), NodeId::new([1; 32])]);
let block1 = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block1.clone()).unwrap();
@ -624,7 +626,11 @@ mod test {
#[test]
// Ensure that approve_block updates highest_voted_view and returns a correct Send.
fn approve_block() {
let mut engine = init(vec![[0; 32], [1; 32], [3; 32]]);
let mut engine = init(vec![
NodeId::new([0; 32]),
NodeId::new([1; 32]),
NodeId::new([3; 32]),
]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block.clone()).unwrap();
@ -649,7 +655,7 @@ mod test {
#[should_panic(expected = "not in")]
// Ensure that approve_block cannot accept not-received blocks.
fn approve_block_not_received() {
let engine = init(vec![[0; 32]]);
let engine = init(vec![NodeId::new([0; 32])]);
let block = next_block(&engine, &engine.genesis_block());
let _ = engine.approve_block(block);
@ -659,7 +665,7 @@ mod test {
#[should_panic(expected = "can't vote for a block in the past")]
// Ensure that approve_block cannot vote blocks in the past.
fn approve_block_in_the_past() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block.clone()).unwrap();
@ -674,7 +680,7 @@ mod test {
#[test]
// Ensure that local_timeout() votes on the current view.
fn local_timeout() {
let mut engine = init(vec![[0; 32], [1; 32]]);
let mut engine = init(vec![NodeId::new([0; 32]), NodeId::new([1; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block).unwrap(); // received but not approved yet
engine = update_leader_selection(&engine);
@ -687,7 +693,7 @@ mod test {
to: engine.overlay().root_committee(),
payload: Payload::Timeout(Timeout {
view: 1,
sender: [0; 32],
sender: NodeId::new([0; 32]),
high_qc: StandardQc {
view: 0, // genesis
id: [0; 32],
@ -701,7 +707,7 @@ mod test {
#[test]
// Ensure that receive_timeout_qc updates current_view, last_view_timeout_qc and local_high_qc.
fn receive_timeout_qc_after_local_timeout() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block).unwrap(); // received but not approved yet
engine = update_leader_selection(&engine);
@ -715,7 +721,7 @@ mod test {
view: 0, // genesis
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
engine = engine.receive_timeout_qc(timeout_qc.clone());
assert_eq!(&engine.local_high_qc, timeout_qc.high_qc());
@ -726,7 +732,7 @@ mod test {
#[test]
// Ensure that receive_timeout_qc works even before local_timeout occurs.
fn receive_timeout_qc_before_local_timeout() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block).unwrap(); // received but not approved yet
engine = update_leader_selection(&engine);
@ -740,7 +746,7 @@ mod test {
view: 0, // genesis
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
engine = engine.receive_timeout_qc(timeout_qc.clone());
assert_eq!(&engine.local_high_qc, timeout_qc.high_qc());
@ -751,7 +757,11 @@ mod test {
#[test]
// Ensure that approve_new_view votes on the new view correctly.
fn approve_new_view() {
let mut engine = init(vec![[0; 32], [1; 32], [2; 32]]);
let mut engine = init(vec![
NodeId::new([0; 32]),
NodeId::new([1; 32]),
NodeId::new([2; 32]),
]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block).unwrap(); // received but not approved yet
engine = update_leader_selection(&engine);
@ -763,7 +773,7 @@ mod test {
view: 0, // genesis
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
engine = engine.receive_timeout_qc(timeout_qc.clone());
assert_eq!(&engine.local_high_qc, timeout_qc.high_qc());
@ -782,7 +792,7 @@ mod test {
to: vec![engine.overlay().next_leader()].into_iter().collect(),
payload: Payload::NewView(NewView {
view: 2,
sender: [0; 32],
sender: NodeId::new([0; 32]),
timeout_qc: timeout_qc.clone(),
high_qc: timeout_qc.high_qc().clone(),
})
@ -793,7 +803,7 @@ mod test {
#[test]
#[should_panic(expected = "can't vote for a new view not bigger than the last timeout_qc")]
fn approve_new_view_not_bigger_than_timeout_qc() {
let mut engine = init(vec![[0; 32]]);
let mut engine = init(vec![NodeId::new([0; 32])]);
let block = next_block(&engine, &engine.genesis_block());
engine = engine.receive_block(block).unwrap(); // received but not approved yet
engine = update_leader_selection(&engine);
@ -805,7 +815,7 @@ mod test {
view: 0, // genesis
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
engine = engine.receive_timeout_qc(timeout_qc1.clone());
assert_eq!(engine.last_view_timeout_qc, Some(timeout_qc1.clone()));
@ -818,7 +828,7 @@ mod test {
view: 0, // genesis
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
engine = engine.receive_timeout_qc(timeout_qc2.clone());
assert_eq!(engine.last_view_timeout_qc, Some(timeout_qc2));

21
consensus-engine/src/overlay/tree_overlay/overlay.rs
View File

@ -206,7 +206,7 @@ mod tests {
#[test]
fn test_carnot_overlay_leader() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let overlay = TreeOverlay::new(TreeOverlaySettings {
nodes: nodes.clone(),
current_leader: nodes[0],
@ -220,7 +220,7 @@ mod tests {
#[test]
fn test_next_leader_is_advance_current_leader() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let mut overlay = TreeOverlay::new(TreeOverlaySettings {
nodes: nodes.clone(),
current_leader: nodes[0],
@ -237,7 +237,7 @@ mod tests {
#[test]
fn test_root_committee() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let overlay = TreeOverlay::new(TreeOverlaySettings {
current_leader: nodes[0],
nodes,
@ -255,7 +255,7 @@ mod tests {
#[test]
fn test_leaf_committees() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let overlay = TreeOverlay::new(TreeOverlaySettings {
current_leader: nodes[0],
nodes,
@ -265,7 +265,7 @@ mod tests {
});
let mut leaf_committees = overlay
.leaf_committees([0; 32])
.leaf_committees(NodeId::new([0; 32]))
.into_iter()
.map(|s| {
let mut vec = s.into_iter().collect::<Vec<_>>();
@ -285,7 +285,7 @@ mod tests {
#[test]
fn test_super_majority_threshold_for_leaf() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let overlay = TreeOverlay::new(TreeOverlaySettings {
current_leader: nodes[0],
nodes,
@ -299,7 +299,7 @@ mod tests {
#[test]
fn test_super_majority_threshold_for_root_member() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let overlay = TreeOverlay::new(TreeOverlaySettings {
current_leader: nodes[0],
nodes,
@ -313,7 +313,7 @@ mod tests {
#[test]
fn test_leader_super_majority_threshold() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let overlay = TreeOverlay::new(TreeOverlaySettings {
nodes: nodes.clone(),
current_leader: nodes[0],
@ -322,6 +322,9 @@ mod tests {
leader: RoundRobin::new(),
});
assert_eq!(overlay.leader_super_majority_threshold([0; 32]), 7);
assert_eq!(
overlay.leader_super_majority_threshold(NodeId::new([0; 32])),
7
);
}
}

6
consensus-engine/src/overlay/tree_overlay/tree.rs
View File

@ -150,7 +150,7 @@ mod tests {
#[test]
fn test_carnot_tree_parenting() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let tree = Tree::new(&nodes, 3);
let root = &tree.inner_committees[0];
@ -163,7 +163,7 @@ mod tests {
#[test]
fn test_carnot_tree_root_parenting() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let tree = Tree::new(&nodes, 3);
let root = &tree.inner_committees[0];
@ -173,7 +173,7 @@ mod tests {
#[test]
fn test_carnot_tree_childs() {
let nodes: Vec<[u8; 32]> = (0..10).map(|i| [i as u8; 32]).collect();
let nodes: Vec<_> = (0..10).map(|i| NodeId::new([i as u8; 32])).collect();
let tree = Tree::new(&nodes, 3);
let root = &tree.inner_committees[0];

17
consensus-engine/src/types.rs
View File

@ -6,9 +6,10 @@ use serde::{Deserialize, Serialize};
mod committee;
pub use committee::{Committee, CommitteeId};
mod node_id;
pub use node_id::NodeId;
pub type View = i64;
pub type NodeId = [u8; 32];
pub type BlockId = [u8; 32];
/// The way the consensus engine communicates with the rest of the system is by returning
@ -118,7 +119,9 @@ impl Block {
id: [0; 32],
view: 0,
parent_qc: Qc::Standard(StandardQc::genesis()),
leader_proof: LeaderProof::LeaderId { leader_id: [0; 32] },
leader_proof: LeaderProof::LeaderId {
leader_id: NodeId::new([0; 32]),
},
}
}
}
@ -225,12 +228,12 @@ mod test {
view: 1,
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
assert_eq!(timeout_qc.view(), 2);
assert_eq!(timeout_qc.high_qc().view, 1);
assert_eq!(timeout_qc.high_qc().id, [0; 32]);
assert_eq!(timeout_qc.sender(), [0; 32]);
assert_eq!(timeout_qc.sender(), NodeId::new([0; 32]));
let timeout_qc = TimeoutQc::new(
2,
@ -238,12 +241,12 @@ mod test {
view: 2,
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
assert_eq!(timeout_qc.view(), 2);
assert_eq!(timeout_qc.high_qc().view, 2);
assert_eq!(timeout_qc.high_qc().id, [0; 32]);
assert_eq!(timeout_qc.sender(), [0; 32]);
assert_eq!(timeout_qc.sender(), NodeId::new([0; 32]));
}
#[test]
@ -257,7 +260,7 @@ mod test {
view: 2,
id: [0; 32],
},
[0; 32],
NodeId::new([0; 32]),
);
}
}

10
consensus-engine/src/types/committee.rs
View File

@ -62,7 +62,7 @@ impl Committee {
) -> digest::generic_array::GenericArray<u8, <D as digest::OutputSizeUser>::OutputSize> {
let mut hasher = D::new();
for member in &self.members {
hasher.update(member);
hasher.update(member.0);
}
hasher.finalize()
}
@ -119,16 +119,16 @@ where
}
}
impl core::iter::FromIterator<[u8; 32]> for Committee {
fn from_iter<T: IntoIterator<Item = [u8; 32]>>(iter: T) -> Self {
impl core::iter::FromIterator<NodeId> for Committee {
fn from_iter<T: IntoIterator<Item = NodeId>>(iter: T) -> Self {
Self {
members: iter.into_iter().collect(),
}
}
}
impl<'a> core::iter::FromIterator<&'a [u8; 32]> for Committee {
fn from_iter<T: IntoIterator<Item = &'a [u8; 32]>>(iter: T) -> Self {
impl<'a> core::iter::FromIterator<&'a NodeId> for Committee {
fn from_iter<T: IntoIterator<Item = &'a NodeId>>(iter: T) -> Self {
Self {
members: iter.into_iter().copied().collect(),
}

46
consensus-engine/src/types/node_id.rs Normal file
View File

@ -0,0 +1,46 @@
#[derive(Clone, Copy, Default, Debug, Eq, PartialEq, Hash, Ord, PartialOrd)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(transparent))]
pub struct NodeId(pub(crate) [u8; 32]);
impl NodeId {
pub const fn new(val: [u8; 32]) -> Self {
Self(val)
}
/// Returns a random node id
#[cfg(any(test, feature = "simulation"))]
pub fn random<R: rand::Rng>(rng: &mut R) -> Self {
let mut bytes = [0u8; 32];
rng.fill_bytes(&mut bytes);
Self(bytes)
}
}
impl From<[u8; 32]> for NodeId {
fn from(id: [u8; 32]) -> Self {
Self(id)
}
}
impl From<&[u8; 32]> for NodeId {
fn from(id: &[u8; 32]) -> Self {
Self(*id)
}
}
impl From<NodeId> for [u8; 32] {
fn from(id: NodeId) -> Self {
id.0
}
}
impl core::fmt::Display for NodeId {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
write!(f, "0x")?;
for v in self.0 {
write!(f, "{:02x}", v)?;
}
Ok(())
}
}

6
consensus-engine/tests/fuzz/ref_state.rs
View File

@ -27,9 +27,11 @@ pub struct ViewEntry {
pub timeout_qcs: HashSet<TimeoutQc>,
}
const LEADER_PROOF: LeaderProof = LeaderProof::LeaderId { leader_id: [0; 32] };
const LEADER_PROOF: LeaderProof = LeaderProof::LeaderId {
leader_id: NodeId::new([0; 32]),
};
const INITIAL_HIGHEST_VOTED_VIEW: View = -1;
const SENDER: NodeId = [0; 32];
const SENDER: NodeId = NodeId::new([0; 32]);
impl ReferenceStateMachine for RefState {
type State = Self;

8
consensus-engine/tests/fuzz/sut.rs
View File

@ -19,15 +19,17 @@ pub struct ConsensusEngineTest {
impl ConsensusEngineTest {
pub fn new() -> Self {
let engine = Carnot::from_genesis(
[0; 32],
NodeId::new([0; 32]),
Block {
view: 0,
id: [0; 32],
parent_qc: Qc::Standard(StandardQc::genesis()),
leader_proof: LeaderProof::LeaderId { leader_id: [0; 32] },
leader_proof: LeaderProof::LeaderId {
leader_id: NodeId::new([0; 32]),
},
},
FlatOverlay::new(Settings {
nodes: vec![[0; 32]],
nodes: vec![NodeId::new([0; 32])],
leader: RoundRobin::default(),
leader_super_majority_threshold: None,
}),

16
nomos-services/consensus/src/lib.rs
View File

@ -11,6 +11,7 @@ use std::pin::Pin;
use std::time::Duration;
// crates
use bls_signatures::PrivateKey;
pub use consensus_engine::NodeId;
use futures::{Stream, StreamExt};
use leader_selection::UpdateableLeaderSelection;
use serde::Deserialize;
@ -31,7 +32,6 @@ use consensus_engine::{
use task_manager::TaskManager;
use nomos_core::block::Block;
use nomos_core::crypto::PublicKey;
use nomos_core::fountain::FountainCode;
use nomos_core::tx::Transaction;
use nomos_core::vote::Tally;
@ -52,8 +52,6 @@ fn default_timeout() -> Duration {
DEFAULT_TIMEOUT
}
// Raw bytes for now, could be a ed25519 public key
pub type NodeId = PublicKey;
// Random seed for each round provided by the protocol
pub type Seed = [u8; 32];
@ -182,9 +180,11 @@ where
id: [0; 32],
view: 0,
parent_qc: Qc::Standard(StandardQc::genesis()),
leader_proof: LeaderProof::LeaderId { leader_id: [0; 32] },
leader_proof: LeaderProof::LeaderId {
leader_id: NodeId::new([0; 32]),
},
};
let mut carnot = Carnot::from_genesis(private_key, genesis, overlay);
let mut carnot = Carnot::from_genesis(NodeId::new(private_key), genesis, overlay);
let adapter = A::new(network_relay).await;
let fountain = F::new(fountain_settings);
let private_key = PrivateKey::new(private_key);
@ -859,7 +859,7 @@ mod tests {
#[test]
fn serde_carnot_info() {
let info = CarnotInfo {
id: [0; 32],
id: NodeId::new([0; 32]),
current_view: 1,
highest_voted_view: -1,
local_high_qc: StandardQc {
@ -875,7 +875,9 @@ mod tests {
view: 0,
id: [0; 32],
}),
leader_proof: LeaderProof::LeaderId { leader_id: [0; 32] },
leader_proof: LeaderProof::LeaderId {
leader_id: NodeId::new([0; 32]),
},
},
)]),
last_view_timeout_qc: None,

2
simulations/Cargo.toml
View File

@ -18,7 +18,7 @@ ctrlc = "3.4"
chrono = { version = "0.4", features = ["serde"] }
crc32fast = "1.3"
crossbeam = { version = "0.8.2", features = ["crossbeam-channel"] }
consensus-engine = { path = "../consensus-engine" }
consensus-engine = { path = "../consensus-engine", features = ["simulation"] }
fixed-slice-deque = "0.1.0-beta2"
futures = "0.3"
humantime = "2.1"

8
simulations/src/bin/app/main.rs
View File

@ -17,7 +17,7 @@ use serde::de::DeserializeOwned;
use simulations::network::behaviour::create_behaviours;
use simulations::network::regions::{create_regions, RegionsData};
use simulations::network::{InMemoryNetworkInterface, Network};
use simulations::node::{Node, NodeId};
use simulations::node::{Node, NodeId, NodeIdExt};
use simulations::output_processors::Record;
use simulations::runner::SimulationRunnerHandle;
use simulations::streaming::{
@ -26,7 +26,7 @@ use simulations::streaming::{
// internal
use simulations::{
node::carnot::CarnotNode, output_processors::OutData, runner::SimulationRunner,
settings::SimulationSettings, util::node_id,
settings::SimulationSettings,
};
mod log;
@ -62,7 +62,9 @@ impl SimulationApp {
.as_secs()
});
let mut rng = SmallRng::seed_from_u64(seed);
let mut node_ids: Vec<NodeId> = (0..simulation_settings.node_count).map(node_id).collect();
let mut node_ids: Vec<NodeId> = (0..simulation_settings.node_count)
.map(NodeId::from_index)
.collect();
node_ids.shuffle(&mut rng);
let regions = create_regions(&node_ids, &mut rng, &simulation_settings.network_settings);

1
simulations/src/lib.rs
View File

@ -7,6 +7,5 @@ pub mod settings;
pub mod streaming;
pub mod warding;
pub mod util;
static START_TIME: once_cell::sync::Lazy<std::time::Instant> =
once_cell::sync::Lazy::new(std::time::Instant::now);

15
simulations/src/network/mod.rs
View File

@ -324,7 +324,10 @@ mod tests {
regions::{Region, RegionsData},
Network, NetworkInterface, NetworkMessage,
};
use crate::{network::NetworkBehaviourKey, node::NodeId, util::node_id};
use crate::{
network::NetworkBehaviourKey,
node::{NodeId, NodeIdExt},
};
use crossbeam::channel::{self, Receiver, Sender};
use std::{collections::HashMap, time::Duration};
@ -368,8 +371,8 @@ mod tests {
#[test]
fn send_receive_messages() {
let node_a = node_id(0);
let node_b = node_id(1);
let node_a = NodeId::from_index(0);
let node_b = NodeId::from_index(1);
let regions = HashMap::from([(Region::Europe, vec![node_a, node_b])]);
let behaviour = HashMap::from([(
@ -417,9 +420,9 @@ mod tests {
#[test]
fn regions_send_receive_messages() {
let node_a = node_id(0);
let node_b = node_id(1);
let node_c = node_id(2);
let node_a = NodeId::from_index(0);
let node_b = NodeId::from_index(1);
let node_c = NodeId::from_index(2);
let regions = HashMap::from([
(Region::Asia, vec![node_a, node_b]),

11
simulations/src/network/regions.rs
View File

@ -1,10 +1,11 @@
use consensus_engine::NodeId;
// std
use rand::{seq::SliceRandom, Rng};
use std::{collections::HashMap, str::FromStr};
// crates
use serde::{Deserialize, Serialize};
// internal
use crate::{network::behaviour::NetworkBehaviour, node::NodeId};
use crate::network::behaviour::NetworkBehaviour;
use super::{NetworkBehaviourKey, NetworkSettings};
@ -156,6 +157,7 @@ pub fn create_regions<R: Rng>(
mod tests {
use std::collections::HashMap;
use consensus_engine::NodeId;
use rand::rngs::mock::StepRng;
use crate::{
@ -163,8 +165,7 @@ mod tests {
regions::{create_regions, Region},
NetworkSettings,
},
node::NodeId,
util::node_id,
node::NodeIdExt,
};
#[test]
@ -203,7 +204,9 @@ mod tests {
let mut rng = StepRng::new(1, 0);
for tcase in test_cases.iter() {
let nodes = (0..tcase.node_count).map(node_id).collect::<Vec<NodeId>>();
let nodes = (0..tcase.node_count)
.map(NodeId::from_index)
.collect::<Vec<NodeId>>();
let available_regions = vec![
Region::NorthAmerica,

9
simulations/src/node/carnot/event_builder.rs
View File

@ -1,5 +1,4 @@
use crate::node::carnot::{messages::CarnotMessage, tally::Tally, timeout::TimeoutHandler};
use crate::util::parse_idx;
use consensus_engine::{
AggregateQc, Carnot, NewView, Overlay, Qc, StandardQc, Timeout, TimeoutQc, View, Vote,
};
@ -69,7 +68,7 @@ impl EventBuilder {
if engine.highest_voted_view() == -1
&& engine.overlay().is_member_of_leaf_committee(self.id)
{
tracing::info!(node = parse_idx(&self.id), "voting genesis",);
tracing::info!(node = %self.id, "voting genesis",);
let genesis = engine.genesis_block();
events.push(Event::Approve {
qc: Qc::Standard(StandardQc {
@ -86,7 +85,7 @@ impl EventBuilder {
CarnotMessage::Proposal(msg) => {
let block = Block::from_bytes(&msg.chunk);
tracing::info!(
node=parse_idx(&self.id),
node=%self.id,
current_view = engine.current_view(),
block_view=block.header().view,
block=?block.header().id,
@ -120,7 +119,7 @@ impl EventBuilder {
};
let Some(qc) = msg.qc.clone() else {
tracing::warn!(node=?parse_idx(&self.id), current_view = engine.current_view(), "received vote without QC");
tracing::warn!(node=%self.id, current_view = engine.current_view(), "received vote without QC");
continue;
};
@ -139,7 +138,7 @@ impl EventBuilder {
.cloned()
{
tracing::info!(
node=parse_idx(&self.id),
node=%self.id,
votes=votes.len(),
current_view = engine.current_view(),
block_view=block.view,

15
simulations/src/node/carnot/mod.rs
View File

@ -19,7 +19,6 @@ use super::{Node, NodeId};
use crate::network::{InMemoryNetworkInterface, NetworkInterface, NetworkMessage};
use crate::node::carnot::event_builder::{CarnotTx, Event};
use crate::node::carnot::message_cache::MessageCache;
use crate::util::parse_idx;
use consensus_engine::overlay::RandomBeaconState;
use consensus_engine::{
Block, BlockId, Carnot, Committee, Overlay, Payload, Qc, StandardQc, TimeoutQc, View, Vote,
@ -353,7 +352,7 @@ impl<L: UpdateableLeaderSelection, O: Overlay<LeaderSelection = L>> Node for Car
Event::Proposal { block } => {
let current_view = self.engine.current_view();
tracing::info!(
node=parse_idx(&self.id),
node=%self.id,
last_committed_view=self.engine.latest_committed_view(),
current_view = current_view,
block_view = block.header().view,
@ -375,7 +374,7 @@ impl<L: UpdateableLeaderSelection, O: Overlay<LeaderSelection = L>> Node for Car
}
}
Err(_) => {
tracing::error!(node = parse_idx(&self.id), current_view = self.engine.current_view(), block_view = block.header().view, block = ?block.header().id, "receive block proposal, but is invalid");
tracing::error!(node = %self.id, current_view = self.engine.current_view(), block_view = block.header().view, block = ?block.header().id, "receive block proposal, but is invalid");
}
}
@ -393,7 +392,7 @@ impl<L: UpdateableLeaderSelection, O: Overlay<LeaderSelection = L>> Node for Car
// So we can just call approve_block
Event::Approve { block, .. } => {
tracing::info!(
node = parse_idx(&self.id),
node = %self.id,
current_view = self.engine.current_view(),
block_view = block.view,
block = ?block.id,
@ -401,7 +400,7 @@ impl<L: UpdateableLeaderSelection, O: Overlay<LeaderSelection = L>> Node for Car
"receive approve message"
);
let (new, out) = self.engine.approve_block(block);
tracing::info!(vote=?out, node=parse_idx(&self.id));
tracing::info!(vote=?out, node=%self.id);
output = vec![Output::Send(out)];
self.engine = new;
}
@ -426,7 +425,7 @@ impl<L: UpdateableLeaderSelection, O: Overlay<LeaderSelection = L>> Node for Car
new_views,
} => {
tracing::info!(
node = parse_idx(&self.id),
node = %self.id,
current_view = self.engine.current_view(),
timeout_view = timeout_qc.view(),
"receive new view message"
@ -437,7 +436,7 @@ impl<L: UpdateableLeaderSelection, O: Overlay<LeaderSelection = L>> Node for Car
}
Event::TimeoutQc { timeout_qc } => {
tracing::info!(
node = parse_idx(&self.id),
node = %self.id,
current_view = self.engine.current_view(),
timeout_view = timeout_qc.view(),
"receive timeout qc message"
@ -467,7 +466,7 @@ impl<L: UpdateableLeaderSelection, O: Overlay<LeaderSelection = L>> Node for Car
}
Event::LocalTimeout => {
tracing::info!(
node = parse_idx(&self.id),
node = %self.id,
current_view = self.engine.current_view(),
"receive local timeout message"
);

118
simulations/src/node/dummy.rs
View File

@ -444,13 +444,12 @@ mod tests {
},
node::{
dummy::{get_child_nodes, get_parent_nodes, get_roles, DummyRole},
Node, NodeId, OverlayState, SharedState, SimulationOverlay, ViewOverlay,
Node, NodeId, NodeIdExt, OverlayState, SharedState, SimulationOverlay, ViewOverlay,
},
overlay::{
tree::{TreeOverlay, TreeSettings},
Overlay,
},
util::node_id,
};
use super::{DummyMessage, DummyNode, Intent, Vote};
@ -522,7 +521,7 @@ mod tests {
.for_each(|leader_id| {
for _ in 0..committee_size {
nodes
.get(&node_id(0))
.get(&NodeId::from_index(0))
.unwrap()
.send_message(*leader_id, DummyMessage::Vote(initial_vote.clone()));
}
@ -544,7 +543,11 @@ mod tests {
let mut network = init_network(&node_ids);
let view = ViewOverlay {
leaders: vec![node_id(0), node_id(1), node_id(2)],
leaders: vec![
NodeId::from_index(0),
NodeId::from_index(1),
NodeId::from_index(2),
],
layout: overlay.layout(&node_ids, &mut rng),
};
let overlay_state = Arc::new(RwLock::new(OverlayState {
@ -562,9 +565,16 @@ mod tests {
let initial_vote = Vote::new(1, Intent::FromRootToLeader);
// Using any node as the sender for initial proposal to leader nodes.
nodes[&node_id(0)].send_message(node_id(0), DummyMessage::Vote(initial_vote.clone()));
nodes[&node_id(0)].send_message(node_id(1), DummyMessage::Vote(initial_vote.clone()));
nodes[&node_id(0)].send_message(node_id(2), DummyMessage::Vote(initial_vote));
nodes[&NodeId::from_index(0)].send_message(
NodeId::from_index(0),
DummyMessage::Vote(initial_vote.clone()),
);
nodes[&NodeId::from_index(0)].send_message(
NodeId::from_index(1),
DummyMessage::Vote(initial_vote.clone()),
);
nodes[&NodeId::from_index(0)]
.send_message(NodeId::from_index(2), DummyMessage::Vote(initial_vote));
network.collect_messages();
for (_, node) in nodes.iter() {
@ -592,15 +602,15 @@ mod tests {
}
// Root and Internal haven't sent their votes yet.
assert!(!nodes[&node_id(0)].state().view_state[&1].vote_sent); // Root
assert!(!nodes[&node_id(1)].state().view_state[&1].vote_sent); // Internal
assert!(!nodes[&node_id(2)].state().view_state[&1].vote_sent); // Internal
assert!(!nodes[&NodeId::from_index(0)].state().view_state[&1].vote_sent); // Root
assert!(!nodes[&NodeId::from_index(1)].state().view_state[&1].vote_sent); // Internal
assert!(!nodes[&NodeId::from_index(2)].state().view_state[&1].vote_sent); // Internal
// Leaves should have thier vote sent.
assert!(nodes[&node_id(3)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(4)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(5)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(6)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(3)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(4)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(5)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(6)].state().view_state[&1].vote_sent); // Leaf
// 3. Internal nodes send vote to root node.
network.dispatch_after(elapsed);
@ -610,15 +620,15 @@ mod tests {
network.collect_messages();
// Root hasn't sent its votes yet.
assert!(!nodes[&node_id(0)].state().view_state[&1].vote_sent); // Root
assert!(!nodes[&NodeId::from_index(0)].state().view_state[&1].vote_sent); // Root
// Internal and leaves should have thier vote sent.
assert!(nodes[&node_id(1)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&node_id(2)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&node_id(3)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(4)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(5)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(6)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(1)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&NodeId::from_index(2)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&NodeId::from_index(3)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(4)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(5)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(6)].state().view_state[&1].vote_sent); // Leaf
// 4. Root node send vote to next view leader nodes.
network.dispatch_after(elapsed);
@ -628,13 +638,13 @@ mod tests {
network.collect_messages();
// Root has sent its votes.
assert!(nodes[&node_id(0)].state().view_state[&1].vote_sent); // Root
assert!(nodes[&node_id(1)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&node_id(2)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&node_id(3)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(4)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(5)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&node_id(6)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(0)].state().view_state[&1].vote_sent); // Root
assert!(nodes[&NodeId::from_index(1)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&NodeId::from_index(2)].state().view_state[&1].vote_sent); // Internal
assert!(nodes[&NodeId::from_index(3)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(4)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(5)].state().view_state[&1].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(6)].state().view_state[&1].vote_sent); // Leaf
// 5. Leaders receive vote and broadcast new Proposal(Block) to all nodes.
network.dispatch_after(elapsed);
@ -662,15 +672,16 @@ mod tests {
}
// Root and Internal haven't sent their votes yet.
assert!(!nodes[&node_id(0)].state().view_state[&2].vote_sent); // Root
assert!(!nodes[&node_id(1)].state().view_state[&2].vote_sent); // Internal
assert!(!nodes[&node_id(2)].state().view_state[&2].vote_sent); // Internal
assert!(!nodes[&NodeId::from_index(0)].state().view_state[&2].vote_sent); // Root
assert!(!nodes[&NodeId::from_index(1)].state().view_state[&2].vote_sent); // Internal
assert!(!nodes[&NodeId::from_index(2)].state().view_state[&2].vote_sent); // Internal
// Leaves should have thier vote sent.
assert!(nodes[&node_id(3)].state().view_state[&2].vote_sent); // Leaf
assert!(nodes[&node_id(4)].state().view_state[&2].vote_sent); // Leaf
assert!(nodes[&node_id(5)].state().view_state[&2].vote_sent); // Leaf
assert!(nodes[&node_id(6)].state().view_state[&2].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(3)].state().view_state[&2].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(4)].state().view_state[&2].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(5)].state().view_state[&2].vote_sent); // Leaf
assert!(nodes[&NodeId::from_index(6)].state().view_state[&2].vote_sent);
// Leaf
}
#[test]
@ -691,7 +702,7 @@ mod tests {
}));
// There are more nodes in the network than in a tree overlay.
let node_ids: Vec<NodeId> = (0..100).map(node_id).collect();
let node_ids: Vec<NodeId> = (0..100).map(NodeId::from_index).collect();
let mut network = init_network(&node_ids);
let overlays = generate_overlays(&node_ids, &overlay, 4, 3, &mut rng);
@ -741,7 +752,7 @@ mod tests {
}));
// There are more nodes in the network than in a tree overlay.
let node_ids: Vec<NodeId> = (0..10000).map(node_id).collect();
let node_ids: Vec<NodeId> = (0..10000).map(NodeId::from_index).collect();
let mut network = init_network(&node_ids);
let overlays = generate_overlays(&node_ids, &overlay, 4, 100, &mut rng);
@ -791,7 +802,7 @@ mod tests {
}));
// There are more nodes in the network than in a tree overlay.
let node_ids: Vec<NodeId> = (0..100000).map(node_id).collect();
let node_ids: Vec<NodeId> = (0..100000).map(NodeId::from_index).collect();
let mut network = init_network(&node_ids);
let overlays = generate_overlays(&node_ids, &overlay, 4, 1000, &mut rng);
@ -831,42 +842,51 @@ mod tests {
(
0,
None,
Some(BTreeSet::from([node_id(1), node_id(2)])),
Some(BTreeSet::from([
NodeId::from_index(1),
NodeId::from_index(2),
])),
vec![DummyRole::Root],
),
(
1,
Some(BTreeSet::from([node_id(0)])),
Some(BTreeSet::from([node_id(3), node_id(4)])),
Some(BTreeSet::from([NodeId::from_index(0)])),
Some(BTreeSet::from([
NodeId::from_index(3),
NodeId::from_index(4),
])),
vec![DummyRole::Internal],
),
(
2,
Some(BTreeSet::from([node_id(0)])),
Some(BTreeSet::from([node_id(5), node_id(6)])),
Some(BTreeSet::from([NodeId::from_index(0)])),
Some(BTreeSet::from([
NodeId::from_index(5),
NodeId::from_index(6),
])),
vec![DummyRole::Internal],
),
(
3,
Some(BTreeSet::from([node_id(1)])),
Some(BTreeSet::from([NodeId::from_index(1)])),
None,
vec![DummyRole::Leaf],
),
(
4,
Some(BTreeSet::from([node_id(1)])),
Some(BTreeSet::from([NodeId::from_index(1)])),
None,
vec![DummyRole::Leaf],
),
(
5,
Some(BTreeSet::from([node_id(2)])),
Some(BTreeSet::from([NodeId::from_index(2)])),
None,
vec![DummyRole::Leaf],
),
(
6,
Some(BTreeSet::from([node_id(2)])),
Some(BTreeSet::from([NodeId::from_index(2)])),
None,
vec![DummyRole::Leader, DummyRole::Leaf],
),
@ -878,12 +898,12 @@ mod tests {
committee_size: 1,
});
let node_ids: Vec<NodeId> = overlay.nodes();
let leaders = vec![node_id(6)];
let leaders = vec![NodeId::from_index(6)];
let layout = overlay.layout(&node_ids, &mut rng);
let view = ViewOverlay { leaders, layout };
for (nid, expected_parents, expected_children, expected_roles) in test_cases {
let node_id = node_id(nid);
let node_id = NodeId::from_index(nid);
let parents = get_parent_nodes(node_id, &view);
let children = get_child_nodes(node_id, &view);
let role = get_roles(node_id, &view, &parents, &children);

24
simulations/src/node/mod.rs
View File

@ -163,7 +163,7 @@ impl Node for usize {
type State = Self;
fn id(&self) -> NodeId {
crate::util::node_id(*self)
NodeId::from_index(*self)
}
fn current_view(&self) -> usize {
@ -179,3 +179,25 @@ impl Node for usize {
self.add_assign(1);
}
}
pub trait NodeIdExt {
fn index(&self) -> usize;
fn from_index(idx: usize) -> Self;
}
impl NodeIdExt for NodeId {
fn index(&self) -> usize {
const SIZE: usize = core::mem::size_of::<usize>();
let mut bytes = [0u8; SIZE];
let src: [u8; 32] = (*self).into();
bytes.copy_from_slice(&src[..SIZE]);
usize::from_be_bytes(bytes)
}
fn from_index(idx: usize) -> Self {
let mut bytes = [0u8; 32];
bytes[..core::mem::size_of::<usize>()].copy_from_slice(&idx.to_be_bytes());
NodeId::new(bytes)
}
}

6
simulations/src/overlay/flat.rs
View File

@ -1,12 +1,12 @@
use consensus_engine::NodeId;
// std
// crates
use rand::prelude::IteratorRandom;
use rand::Rng;
// internal
use super::Overlay;
use crate::node::NodeId;
use crate::node::NodeIdExt;
use crate::overlay::{Committee, Layout};
use crate::util::node_id;
pub struct FlatOverlay;
impl FlatOverlay {
@ -23,7 +23,7 @@ impl Default for FlatOverlay {
impl Overlay for FlatOverlay {
fn nodes(&self) -> Vec<NodeId> {
(0..10).map(node_id).collect()
(0..10).map(NodeId::from_index).collect()
}
fn leaders<R: Rng>(

23
simulations/src/overlay/tree.rs
View File

@ -1,14 +1,12 @@
// std
use consensus_engine::NodeId;
use std::collections::HashMap;
// crates
use rand::seq::IteratorRandom;
use serde::{Deserialize, Serialize};
// internal
use super::{Committee, Layout, Overlay};
use crate::{
node::{CommitteeId, NodeId},
util::node_id,
};
use crate::node::{CommitteeId, NodeIdExt};
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub enum TreeType {
@ -48,7 +46,7 @@ impl TreeOverlay {
}
fn build_full_binary_tree<R: rand::Rng>(
node_ids: &[NodeId],
node_id: &[NodeId],
rng: &mut R,
settings: &TreeSettings,
) -> Layout {
@ -56,14 +54,14 @@ impl TreeOverlay {
// For full binary tree to be formed from existing nodes
// a certain unique node count needs to be provided.
assert!(properties.node_count <= node_ids.len());
assert!(properties.node_count <= node_id.len());
let mut committees = HashMap::new();
let mut parents = HashMap::new();
let mut children = HashMap::new();
let mut layers = HashMap::new();
for (committee_id, nodes) in node_ids
for (committee_id, nodes) in node_id
.iter()
.choose_multiple(rng, properties.node_count)
.chunks(settings.committee_size)
@ -108,7 +106,7 @@ impl TreeOverlay {
impl Overlay for TreeOverlay {
fn nodes(&self) -> Vec<NodeId> {
let properties = get_tree_properties(&self.settings);
(0..properties.node_count).map(node_id).collect()
(0..properties.node_count).map(NodeId::from_index).collect()
}
fn leaders<R: rand::Rng>(
@ -154,7 +152,6 @@ fn get_layer(id: usize) -> CommitteeId {
#[cfg(test)]
mod tests {
use crate::util::node_id;
use super::*;
use rand::rngs::mock::StepRng;
@ -229,13 +226,13 @@ mod tests {
let root_nodes = &layout.committees[&CommitteeId::new(0)].nodes;
assert_eq!(root_nodes.len(), 10);
assert_eq!(root_nodes.first(), Some(&node_id(0)));
assert_eq!(root_nodes.last(), Some(&node_id(9)));
assert_eq!(root_nodes.first(), Some(&NodeId::from_index(0)));
assert_eq!(root_nodes.last(), Some(&NodeId::from_index(9)));
let last_nodes = &layout.committees[&CommitteeId::new(1022)].nodes;
assert_eq!(last_nodes.len(), 10);
assert_eq!(last_nodes.first(), Some(&node_id(10220)));
assert_eq!(last_nodes.last(), Some(&node_id(10229)));
assert_eq!(last_nodes.first(), Some(&NodeId::from_index(10220)));
assert_eq!(last_nodes.last(), Some(&NodeId::from_index(10229)));
}
#[test]

8
simulations/src/runner/glauber_runner.rs
View File

@ -1,7 +1,6 @@
use crate::node::{Node, NodeId};
use crate::node::{Node, NodeId, NodeIdExt};
use crate::output_processors::Record;
use crate::runner::SimulationRunner;
use crate::util::{node_id, parse_idx};
use crate::warding::SimulationState;
use crossbeam::channel::bounded;
use crossbeam::select;
@ -38,7 +37,8 @@ where
let mut inner_runner = runner.inner;
let nodes = runner.nodes;
let nodes_remaining: BTreeSet<NodeId> = (0..nodes.read().len()).map(node_id).collect();
let nodes_remaining: BTreeSet<NodeId> =
(0..nodes.read().len()).map(NodeId::from_index).collect();
let iterations: Vec<_> = (0..maximum_iterations).collect();
let (stop_tx, stop_rx) = bounded(1);
let p = runner.producer.clone();
@ -61,7 +61,7 @@ where
{
let mut shared_nodes = nodes.write();
let node: &mut N = shared_nodes
.get_mut(parse_idx(&node_id))
.get_mut(node_id.index())
.expect("Node should be present");
node.step(elapsed);
}

5
simulations/src/runner/layered_runner.rs
View File

@ -40,10 +40,9 @@ use rand::prelude::{IteratorRandom, SliceRandom};
use rand::rngs::SmallRng;
use serde::Serialize;
// internal
use crate::node::{Node, NodeId};
use crate::node::{Node, NodeId, NodeIdExt};
use crate::output_processors::Record;
use crate::runner::SimulationRunner;
use crate::util::parse_idx;
use crate::warding::SimulationState;
use super::SimulationRunnerHandle;
@ -98,7 +97,7 @@ where
{
let mut shared_nodes = nodes.write();
let node: &mut N = shared_nodes
.get_mut(parse_idx(&node_id))
.get_mut(node_id.index())
.expect("Node should be present");
let prev_view = node.current_view();
node.step(elapsed);

7
simulations/src/runner/sync_runner.rs
View File

@ -75,14 +75,13 @@ mod tests {
},
node::{
dummy::{DummyMessage, DummyNode},
Node, NodeId, OverlayState, SharedState, ViewOverlay,
Node, NodeId, NodeIdExt, OverlayState, SharedState, ViewOverlay,
},
output_processors::OutData,
overlay::{tree::TreeOverlay, Overlay, SimulationOverlay},
runner::SimulationRunner,
settings::SimulationSettings,
streaming::StreamProducer,
util::node_id,
};
use crossbeam::channel;
use parking_lot::RwLock;
@ -131,7 +130,7 @@ mod tests {
};
let mut rng = StepRng::new(1, 0);
let node_ids: Vec<NodeId> = (0..settings.node_count).map(node_id).collect();
let node_ids: Vec<NodeId> = (0..settings.node_count).map(NodeId::from_index).collect();
let overlay = TreeOverlay::new(settings.overlay_settings.clone().try_into().unwrap());
let mut network = init_network(&node_ids);
let view = ViewOverlay {
@ -166,7 +165,7 @@ mod tests {
};
let mut rng = StepRng::new(1, 0);
let node_ids: Vec<NodeId> = (0..settings.node_count).map(node_id).collect();
let node_ids: Vec<NodeId> = (0..settings.node_count).map(NodeId::from_index).collect();
let overlay = TreeOverlay::new(settings.overlay_settings.clone().try_into().unwrap());
let mut network = init_network(&node_ids);
let view = ViewOverlay {

9
simulations/src/streaming/io.rs
View File

@ -118,10 +118,9 @@ mod tests {
regions::{Region, RegionsData},
Network, NetworkBehaviourKey,
},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId, NodeIdExt},
output_processors::OutData,
runner::SimulationRunner,
util::node_id,
warding::SimulationState,
};
@ -153,7 +152,7 @@ mod tests {
};
let nodes = (0..6)
.map(|idx| DummyStreamingNode::new(node_id(idx), ()))
.map(|idx| DummyStreamingNode::new(NodeId::from_index(idx), ()))
.collect::<Vec<_>>();
let network = Network::new(RegionsData {
regions: (0..6)
@ -167,7 +166,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(region, vec![node_id(idx)])
(region, vec![NodeId::from_index(idx)])
})
.collect(),
node_region: (0..6)
@ -181,7 +180,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(node_id(idx), region)
(NodeId::from_index(idx), region)
})
.collect(),
region_network_behaviour: (0..6)

9
simulations/src/streaming/naive.rs
View File

@ -122,10 +122,9 @@ mod tests {
regions::{Region, RegionsData},
Network, NetworkBehaviourKey,
},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId, NodeIdExt},
output_processors::OutData,
runner::SimulationRunner,
util::node_id,
warding::SimulationState,
};
@ -158,7 +157,7 @@ mod tests {
};
let nodes = (0..6)
.map(|idx| DummyStreamingNode::new(node_id(idx), ()))
.map(|idx| DummyStreamingNode::new(NodeId::from_index(idx), ()))
.collect::<Vec<_>>();
let network = Network::new(RegionsData {
regions: (0..6)
@ -172,7 +171,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(region, vec![node_id(idx)])
(region, vec![NodeId::from_index(idx)])
})
.collect(),
node_region: (0..6)
@ -186,7 +185,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(node_id(idx), region)
(NodeId::from_index(idx), region)
})
.collect(),
region_network_behaviour: (0..6)

9
simulations/src/streaming/runtime_subscriber.rs
View File

@ -107,10 +107,9 @@ mod tests {
regions::{Region, RegionsData},
Network, NetworkBehaviourKey,
},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId, NodeIdExt},
output_processors::OutData,
runner::SimulationRunner,
util::node_id,
warding::SimulationState,
};
@ -143,7 +142,7 @@ mod tests {
};
let nodes = (0..6)
.map(|idx| DummyStreamingNode::new(node_id(idx), ()))
.map(|idx| DummyStreamingNode::new(NodeId::from_index(idx), ()))
.collect::<Vec<_>>();
let network = Network::new(RegionsData {
regions: (0..6)
@ -157,7 +156,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(region, vec![node_id(idx)])
(region, vec![NodeId::from_index(idx)])
})
.collect(),
node_region: (0..6)
@ -171,7 +170,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(node_id(idx), region)
(NodeId::from_index(idx), region)
})
.collect(),
region_network_behaviour: (0..6)

9
simulations/src/streaming/settings_subscriber.rs
View File

@ -107,10 +107,9 @@ mod tests {
regions::{Region, RegionsData},
Network, NetworkBehaviourKey,
},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId},
node::{dummy_streaming::DummyStreamingNode, Node, NodeId, NodeIdExt},
output_processors::OutData,
runner::SimulationRunner,
util::node_id,
warding::SimulationState,
};
@ -143,7 +142,7 @@ mod tests {
};
let nodes = (0..6)
.map(|idx| DummyStreamingNode::new(node_id(idx), ()))
.map(|idx| DummyStreamingNode::new(NodeId::from_index(idx), ()))
.collect::<Vec<_>>();
let network = Network::new(RegionsData {
regions: (0..6)
@ -157,7 +156,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(region, vec![node_id(idx)])
(region, vec![NodeId::from_index(idx)])
})
.collect(),
node_region: (0..6)
@ -171,7 +170,7 @@ mod tests {
5 => Region::Australia,
_ => unreachable!(),
};
(node_id(idx), region)
(NodeId::from_index(idx), region)
})
.collect(),
region_network_behaviour: (0..6)

20
simulations/src/util.rs
View File

@ -1,20 +0,0 @@
/// Create a random node id.
///
/// The format is:
///
/// [0..4]: node index in big endian
/// [4..32]: zeros
pub fn node_id(id: usize) -> consensus_engine::NodeId {
let mut bytes = [0; 32];
bytes[..4].copy_from_slice((id as u32).to_be_bytes().as_ref());
bytes
}
/// Parse the original index from NodeId
pub(crate) fn parse_idx(id: &consensus_engine::NodeId) -> usize {
let mut bytes = [0; 4];
bytes.copy_from_slice(&id[..4]);
u32::from_be_bytes(bytes) as usize
}
pub(crate) mod millis_duration {}

12
tests/src/nodes/nomos.rs
View File

@ -5,6 +5,7 @@ use std::time::Duration;
// internal
use crate::{get_available_port, Node, SpawnConfig, RNG};
use consensus_engine::overlay::{RoundRobin, Settings};
use consensus_engine::NodeId;
use nomos_consensus::{CarnotInfo, CarnotSettings};
use nomos_http::backends::axum::AxumBackendSettings;
use nomos_log::{LoggerBackend, LoggerFormat};
@ -154,7 +155,14 @@ impl Node for NomosNode {
}
let mut configs = ids
.iter()
.map(|id| create_node_config(ids.clone(), *id, threshold, timeout))
.map(|id| {
create_node_config(
ids.iter().copied().map(NodeId::new).collect(),
*id,
threshold,
timeout,
)
})
.collect::<Vec<_>>();
let mut nodes = vec![Self::spawn(configs.swap_remove(0)).await];
let listening_addr = nodes[0].get_listening_address().await;
@ -185,7 +193,7 @@ impl Node for NomosNode {
}
fn create_node_config(
nodes: Vec<[u8; 32]>,
nodes: Vec<NodeId>,
private_key: [u8; 32],
threshold: Fraction,
timeout: Duration,