poanetwork
/
hbbft
mirror of https://github.com/poanetwork/hbbft.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add bandwidth and CPU to the simulation.

This commit is contained in:
Andreas Fackler 2018-05-31 12:52:02 +02:00
parent 5e38582828
commit f970272d3e
7 changed files with 180 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.toml
View File

@ -41,3 +41,4 @@ required-features = [ "serialization-protobuf" ]
[[example]]
name = "simulation"
required-features = [ "serialization-serde" ]

2
README.md
View File

@ -13,7 +13,7 @@ or incomplete.
An example is included to run a simulation of a network:
$ cargo run --example simulation -- --nodes 10 --faulty 1 --txs 1000 --batch 100
$ cargo run --example simulation --features=serialization-serde -- -h
# Building

250
examples/simulation.rs
View File

@ -1,23 +1,27 @@
extern crate bincode;
extern crate colored;
extern crate docopt;
extern crate hbbft;
extern crate itertools;
extern crate rand;
#[macro_use(Deserialize)]
extern crate serde;
#[macro_use(Deserialize, Serialize)]
extern crate serde_derive;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::fmt::Debug;
use std::rc::Rc;
use std::time::{Duration, Instant};
use std::{cmp, u64};
use colored::*;
use docopt::Docopt;
use itertools::Itertools;
use rand::Rng;
use serde::de::DeserializeOwned;
use serde::Serialize;
use hbbft::honey_badger::{self, Batch, HoneyBadger};
use hbbft::messaging::{DistAlgorithm, NetworkInfo, Target, TargetedMessage};
use hbbft::honey_badger::{Batch, HoneyBadger};
use hbbft::messaging::{DistAlgorithm, NetworkInfo, Target};
const VERSION: &str = env!("CARGO_PKG_VERSION");
const USAGE: &str = "
@ -29,13 +33,16 @@ Usage:
benchmark --version
Options:
-h, --help Show this message.
--version Show the version of hbbft.
-n <n>, --nodes <n> The total number of nodes [default: 10]
-f <f>, --faulty <f> The number of faulty nodes [default: 0]
-t <txs>, --txs <txs> The number of transactions to process [default: 1000]
-b <b>, --batch <b> The batch size, i.e. txs per epoch [default: 100]
-l <lag>, --lag <lag> The network lag between sending and receiving [default: 100]
-h, --help Show this message.
--version Show the version of hbbft.
-n <n>, --nodes <n> The total number of nodes [default: 10]
-f <f>, --faulty <f> The number of faulty nodes [default: 0]
-t <txs>, --txs <txs> The number of transactions to process [default: 1000]
-b <b>, --batch <b> The batch size, i.e. txs per epoch [default: 100]
-l <lag>, --lag <lag> The network lag between sending and receiving [default: 100]
--bw <bw> The bandwidth, in Kbit/s [default: 10000]
--cpu <cpu> The CPU time, in percent of this machine's [default: 100]
--tx-size <size> The size of a transaction, in bytes [default: 10]
";
#[derive(Deserialize)]
@ -45,18 +52,32 @@ struct Args {
flag_txs: usize,
flag_b: usize,
flag_lag: u64,
flag_bw: u32,
flag_cpu: u32,
flag_tx_size: usize,
}
/// A node identifier. In the simulation, nodes are simply numbered.
#[derive(Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Clone, Copy)]
#[derive(Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Clone, Copy)]
pub struct NodeUid(pub usize);
/// A message with a sender and the timestamp of arrival.
#[derive(Eq, PartialEq, Ord, PartialOrd, Debug)]
/// A transaction.
#[derive(Serialize, Deserialize, Eq, PartialEq, Hash, Ord, PartialOrd, Debug, Clone)]
pub struct Transaction(pub Vec<u8>);
impl Transaction {
fn new(len: usize) -> Transaction {
Transaction(rand::thread_rng().gen_iter().take(len).collect())
}
}
/// A serialized message with a sender and the timestamp of arrival.
#[derive(Eq, PartialEq, Debug)]
struct TimestampedMessage<D: DistAlgorithm> {
time: u64,
time: Duration,
sender_id: D::NodeUid,
message: D::Message,
target: Target<D::NodeUid>,
message: Vec<u8>,
}
impl<D: DistAlgorithm> Clone for TimestampedMessage<D>
@ -67,57 +88,114 @@ where
TimestampedMessage {
time: self.time,
sender_id: self.sender_id.clone(),
target: self.target.clone(),
message: self.message.clone(),
}
}
}
/// Performance parameters of a node's hardware and Internet connection. For simplicity, only the
/// sender's lag and bandwidth are taken into account. (I.e. infinite downstream, limited
/// upstream.)
#[derive(Clone, Copy)]
pub struct HwQuality {
/// The network latency. This is added once for every message.
latency: Duration,
/// The inverse bandwidth, in time per byte.
inv_bw: Duration,
/// The CPU time multiplier: how much slower, in percent, is this node than your computer?
cpu_factor: u32,
}
/// A "node" running an instance of the algorithm `D`.
pub struct TestNode<D: DistAlgorithm> {
/// This node's own ID.
id: D::NodeUid,
/// The instance of the broadcast algorithm.
algo: D,
/// The number of (virtual) milliseconds for which this node has already been simulated.
time: u64,
/// The duration for which this node's CPU has already been simulated.
time: Duration,
/// The time when this node last sent data over the network.
sent_time: Duration,
/// Incoming messages from other nodes that this node has not yet handled, with timestamps.
queue: VecDeque<TimestampedMessage<D>>,
in_queue: VecDeque<TimestampedMessage<D>>,
/// Outgoing messages to other nodes, with timestamps.
out_queue: VecDeque<TimestampedMessage<D>>,
/// The values this node has output so far, with timestamps.
outputs: Vec<(u64, D::Output)>,
outputs: Vec<(Duration, D::Output)>,
/// The number of messages this node has handled so far.
message_count: usize,
/// The hardware and network quality of this node.
hw_quality: HwQuality,
}
impl<D: DistAlgorithm> TestNode<D> {
impl<D: DistAlgorithm> TestNode<D>
where
D::Message: Serialize + DeserializeOwned,
{
/// Creates a new test node with the given broadcast instance.
fn new(mut algo: D) -> TestNode<D> {
let outputs = algo.output_iter().map(|out| (0, out)).collect();
TestNode {
fn new(algo: D, hw_quality: HwQuality) -> TestNode<D> {
let mut node = TestNode {
id: algo.our_id().clone(),
algo,
time: 0,
queue: VecDeque::new(),
outputs,
time: Duration::default(),
sent_time: Duration::default(),
in_queue: VecDeque::new(),
out_queue: VecDeque::new(),
outputs: Vec::new(),
message_count: 0,
}
hw_quality,
};
node.send_output_and_msgs();
node
}
/// Handles the first message in the node's queue.
fn handle_message(&mut self) {
let ts_msg = self.queue.pop_front().expect("message not found");
let ts_msg = self.in_queue.pop_front().expect("message not found");
self.time = cmp::max(self.time, ts_msg.time);
self.message_count += 1;
let start = Instant::now();
let msg = bincode::deserialize::<D::Message>(&ts_msg.message).expect("deserialize");
self.algo
.handle_message(&ts_msg.sender_id, ts_msg.message)
.handle_message(&ts_msg.sender_id, msg)
.expect("handling message");
self.time += start.elapsed() * self.hw_quality.cpu_factor / 100;
self.send_output_and_msgs()
}
/// Handles the algorithm's output and messages.
fn send_output_and_msgs(&mut self) {
let start = Instant::now();
let out_msgs: Vec<_> = self
.algo
.message_iter()
.map(|msg| {
(
msg.target,
bincode::serialize(&msg.message).expect("serialize"),
)
})
.collect();
self.time += start.elapsed() * self.hw_quality.cpu_factor / 100;
let time = self.time;
self.outputs
.extend(self.algo.output_iter().map(|out| (time, out)));
self.sent_time = cmp::max(self.time, self.sent_time);
for (target, message) in out_msgs {
self.sent_time += self.hw_quality.inv_bw * message.len() as u32;
self.out_queue.push_back(TimestampedMessage {
time: self.sent_time + self.hw_quality.latency,
sender_id: self.id.clone(),
target,
message,
});
}
}
/// Returns the time when the next message can be handled.
fn next_event_time(&self) -> Option<u64> {
match self.queue.front() {
fn next_event_time(&self) -> Option<Duration> {
match self.in_queue.front() {
None => None,
Some(ts_msg) => Some(cmp::max(ts_msg.time, self.time)),
}
@ -127,62 +205,71 @@ impl<D: DistAlgorithm> TestNode<D> {
fn message_count(&self) -> usize {
self.message_count
}
/// Adds a message into the incoming queue.
fn add_message(&mut self, msg: TimestampedMessage<D>) {
match self.in_queue.iter().position(|other| other.time > msg.time) {
None => self.in_queue.push_back(msg),
Some(i) => self.in_queue.insert(i, msg),
}
}
}
/// A collection of `TestNode`s representing a network.
pub struct TestNetwork<D: DistAlgorithm> {
nodes: BTreeMap<D::NodeUid, TestNode<D>>,
/// The delay between a message being sent and received.
net_lag: u64,
}
impl<D: DistAlgorithm<NodeUid = NodeUid>> TestNetwork<D>
where
D::Message: Clone,
D::Message: Serialize + DeserializeOwned + Clone,
{
/// Creates a new network with `good_num` good nodes, and `dead_num` dead nodes.
pub fn new<F>(good_num: usize, dead_num: usize, new_algo: F, net_lag: u64) -> TestNetwork<D>
pub fn new<F>(
good_num: usize,
dead_num: usize,
new_algo: F,
hw_quality: HwQuality,
) -> TestNetwork<D>
where
F: Fn(NodeUid, BTreeSet<NodeUid>) -> D,
{
let node_ids: BTreeSet<NodeUid> = (0..(good_num + dead_num)).map(NodeUid).collect();
let new_node_by_id = |id: NodeUid| (id, TestNode::new(new_algo(id, node_ids.clone())));
let new_node_by_id = |id: NodeUid| {
(
id,
TestNode::new(new_algo(id, node_ids.clone()), hw_quality),
)
};
let mut network = TestNetwork {
nodes: (0..good_num).map(NodeUid).map(new_node_by_id).collect(),
net_lag,
};
let mut initial_msgs: Vec<(D::NodeUid, u64, Vec<_>)> = Vec::new();
for (id, node) in &mut network.nodes {
initial_msgs.push((*id, node.time, node.algo.message_iter().collect()));
}
for (id, time, ts_msgs) in initial_msgs {
network.dispatch_messages(id, time, ts_msgs);
}
let initial_msgs: Vec<_> = network
.nodes
.values_mut()
.flat_map(|node| node.out_queue.drain(..))
.collect();
network.dispatch_messages(initial_msgs);
network
}
/// Pushes the messages into the queues of the corresponding recipients.
fn dispatch_messages<Q>(&mut self, sender_id: NodeUid, time: u64, msgs: Q)
fn dispatch_messages<Q>(&mut self, msgs: Q)
where
Q: IntoIterator<Item = TargetedMessage<D::Message, NodeUid>> + Debug,
Q: IntoIterator<Item = TimestampedMessage<D>>,
{
for msg in msgs {
let ts_msg = TimestampedMessage {
sender_id,
time: time + self.net_lag,
message: msg.message,
};
match msg.target {
for ts_msg in msgs {
match ts_msg.target {
Target::All => {
for node in self.nodes.values_mut() {
if node.id != sender_id {
node.queue.push_back(ts_msg.clone())
if node.id != ts_msg.sender_id {
node.add_message(ts_msg.clone())
}
}
}
Target::Node(to_id) => {
if let Some(node) = self.nodes.get_mut(&to_id) {
node.queue.push_back(ts_msg);
node.add_message(ts_msg);
}
}
}
@ -207,9 +294,9 @@ where
let msgs: Vec<_> = {
let node = self.nodes.get_mut(&next_id).unwrap();
node.handle_message();
node.algo.message_iter().collect()
node.out_queue.drain(..).collect()
};
self.dispatch_messages(next_id, min_time, msgs);
self.dispatch_messages(msgs);
next_id
}
@ -222,12 +309,19 @@ where
/// The timestamped batches for a particular epoch that have already been output.
#[derive(Clone, Default)]
struct EpochInfo {
nodes: BTreeMap<NodeUid, (u64, Batch<usize>)>,
nodes: BTreeMap<NodeUid, (Duration, Batch<Transaction>)>,
}
impl EpochInfo {
/// Adds a batch to this epoch. Prints information if the epoch is complete.
fn add(&mut self, id: NodeUid, time: u64, batch: &Batch<usize>, node_num: usize, msgs: usize) {
fn add(
&mut self,
id: NodeUid,
time: Duration,
batch: &Batch<Transaction>,
node_num: usize,
msgs: usize,
) {
if self.nodes.contains_key(&id) {
return;
}
@ -248,7 +342,7 @@ impl EpochInfo {
println!(
"{:>5} {:6} {:5} {:7}",
batch.epoch.to_string().cyan(),
max_t,
max_t.as_secs() * 1000 + max_t.subsec_nanos() as u64 / 1_000_000,
txs,
msgs,
);
@ -256,26 +350,22 @@ impl EpochInfo {
}
/// Proposes `num_txs` values and expects nodes to output and order them.
fn simulate_honey_badger(mut network: TestNetwork<HoneyBadger<usize, NodeUid>>, num_txs: usize) {
fn simulate_honey_badger(
mut network: TestNetwork<HoneyBadger<Transaction, NodeUid>>,
num_txs: usize,
) {
// Returns `true` if the node has not output all transactions yet.
// If it has, and has advanced another epoch, it clears all messages for later epochs.
let node_busy = |node: &mut TestNode<HoneyBadger<usize, NodeUid>>| {
let mut min_missing = 0;
let node_busy = |node: &mut TestNode<HoneyBadger<Transaction, NodeUid>>| {
let mut missing = num_txs;
for &(_, ref batch) in &node.outputs {
for tx in &batch.transactions {
if *tx >= min_missing {
min_missing = tx + 1;
}
}
missing -= &batch.transactions.len();
}
if min_missing < num_txs {
if missing > 0 {
return true;
}
if node.outputs.last().unwrap().1.transactions.is_empty() {
let last = node.outputs.last().unwrap().1.epoch;
node.queue.retain(|ts_msg| match ts_msg.message {
honey_badger::Message::CommonSubset(e, _) => e < last,
});
node.in_queue.clear();
}
false
};
@ -318,10 +408,16 @@ fn main() {
println!("Network lag: {}", args.flag_lag);
println!();
let num_good_nodes = args.flag_n - args.flag_f;
let txs = (0..args.flag_txs).map(|_| Transaction::new(args.flag_tx_size));
let new_honey_badger = |id: NodeUid, all_ids: BTreeSet<NodeUid>| {
let netinfo = Rc::new(NetworkInfo::new(id, all_ids));
HoneyBadger::new(netinfo, args.flag_b, 0..args.flag_txs).expect("Instantiate honey_badger")
HoneyBadger::new(netinfo, args.flag_b, txs.clone()).expect("Instantiate honey_badger")
};
let network = TestNetwork::new(num_good_nodes, args.flag_f, new_honey_badger, args.flag_lag);
let hw_quality = HwQuality {
latency: Duration::from_millis(args.flag_lag),
inv_bw: Duration::new(0, 8_000_000 / args.flag_bw),
cpu_factor: args.flag_cpu,
};
let network = TestNetwork::new(num_good_nodes, args.flag_f, new_honey_badger, hw_quality);
simulate_honey_badger(network, args.flag_txs);
}

2
src/agreement/bin_values.rs
View File

@ -3,7 +3,7 @@ use std::mem::replace;
/// A lattice-valued description of the state of `bin_values`, essentially the same as the set of
/// subsets of `bool`.
#[cfg_attr(feature = "serialization-serde", derive(Serialize))]
#[cfg_attr(feature = "serialization-serde", derive(Serialize, Deserialize))]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum BinValues {
None,

4
src/agreement/mod.rs
View File

@ -23,7 +23,7 @@ error_chain!{
}
}
#[cfg_attr(feature = "serialization-serde", derive(Serialize))]
#[cfg_attr(feature = "serialization-serde", derive(Serialize, Deserialize))]
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum AgreementContent {
/// `BVal` message.
@ -35,7 +35,7 @@ pub enum AgreementContent {
}
/// Messages sent during the binary Byzantine agreement stage.
#[cfg_attr(feature = "serialization-serde", derive(Serialize))]
#[cfg_attr(feature = "serialization-serde", derive(Serialize, Deserialize))]
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct AgreementMessage {
pub epoch: u32,

2
src/common_subset.rs
View File

@ -33,7 +33,7 @@ error_chain!{
type ProposedValue = Vec<u8>;
/// Message from Common Subset to remote nodes.
#[cfg_attr(feature = "serialization-serde", derive(Serialize))]
#[cfg_attr(feature = "serialization-serde", derive(Serialize, Deserialize))]
#[derive(Clone, Debug)]
pub enum Message<NodeUid> {
/// A message for the broadcast algorithm concerning the set element proposed by the given node.

2
src/honey_badger.rs
View File

@ -260,7 +260,7 @@ pub struct Batch<T> {
}
/// A message sent to or received from another node's Honey Badger instance.
#[cfg_attr(feature = "serialization-serde", derive(Serialize))]
#[cfg_attr(feature = "serialization-serde", derive(Serialize, Deserialize))]
#[derive(Debug, Clone)]
pub enum Message<N> {
/// A message belonging to the common subset algorithm in the given epoch.