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

Merge pull request #30 from poanetwork/vk-dist-algo-agreement 

The BA module now defines a DistAlgorithm instance
This commit is contained in:
Andreas Fackler 2018-05-17 12:51:47 +02:00 committed by GitHub
parent 4b0eebf299 a0627b2550
commit 31ec37630f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 180 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
src/agreement.rs
View File

@ -5,11 +5,7 @@ use std::collections::{BTreeSet, HashMap, VecDeque};
use std::fmt::Debug;
use std::hash::Hash;
/// Type of output from the Agreement message handler. The first component is
/// the value on which the Agreement has decided, also called "output" in the
/// HoneyadgerBFT paper. The second component is a queue of messages to be sent
/// to remote nodes as a result of handling the incomming message.
type AgreementOutput = (Option<bool>, VecDeque<AgreementMessage>);
use messaging::{DistAlgorithm, Target, TargetedMessage};
/// Messages sent during the binary Byzantine agreement stage.
#[cfg_attr(feature = "serialization-serde", derive(Serialize))]
@ -42,12 +38,74 @@ pub struct Agreement<NodeUid> {
/// and then never changed. That is, no instance of Binary Agreement can
/// decide on two different values of output.
output: Option<bool>,
/// A permanent, latching copy of the output value. This copy is required because `output` can
/// be consumed using `DistAlgorithm::next_output` immediately after the instance finishing to
/// handle a message, in which case it would otherwise be unknown whether the output value was
/// ever there at all. While the output value will still be required in a later epoch to decide
/// the termination state.
decision: Option<bool>,
/// Termination flag. The Agreement instance doesn't terminate immediately
/// upon deciding on the agreed value. This is done in order to help other
/// nodes decide despite asynchrony of communication. Once the instance
/// determines that all the remote nodes have reached agreement, it sets the
/// `terminated` flag and accepts no more incoming messages.
terminated: bool,
/// The outgoing message queue.
messages: VecDeque<AgreementMessage>,
}
impl<NodeUid: Clone + Debug + Eq + Hash + Ord> DistAlgorithm for Agreement<NodeUid> {
type NodeUid = NodeUid;
type Input = bool;
type Output = bool;
type Message = AgreementMessage;
type Error = Error;
fn input(&mut self, input: Self::Input) -> Result<(), Self::Error> {
self.set_input(input)
}
/// Receive input from a remote node.
fn handle_message(
&mut self,
sender_id: &Self::NodeUid,
message: Self::Message,
) -> Result<(), Self::Error> {
match message {
// The algorithm instance has already terminated.
_ if self.terminated => Err(Error::Terminated),
AgreementMessage::BVal(epoch, b) if epoch == self.epoch => {
self.handle_bval(sender_id, b)
}
AgreementMessage::Aux(epoch, b) if epoch == self.epoch => self.handle_aux(sender_id, b),
// Epoch does not match. Ignore the message.
_ => Ok(()),
}
}
/// Take the next Agreement message for multicast to all other nodes.
fn next_message(&mut self) -> Option<TargetedMessage<Self::Message, Self::NodeUid>> {
self.messages
.pop_front()
.map(|msg| Target::All.message(msg))
}
/// Consume the output. Once consumed, the output stays `None` forever.
fn next_output(&mut self) -> Option<Self::Output> {
self.output.take()
}
/// Whether the algorithm has terminated.
fn terminated(&self) -> bool {
self.terminated
}
fn our_id(&self) -> &Self::NodeUid {
&self.uid
}
}
impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
@ -65,22 +123,15 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
received_aux: HashMap::new(),
estimated: None,
output: None,
decision: None,
terminated: false,
messages: VecDeque::new(),
}
}
pub fn our_id(&self) -> &NodeUid {
&self.uid
}
/// Algorithm has terminated.
pub fn terminated(&self) -> bool {
self.terminated
}
/// Sets the input value for agreement.
pub fn set_input(&mut self, input: bool) -> Result<AgreementMessage, Error> {
if self.epoch != 0 {
pub fn set_input(&mut self, input: bool) -> Result<(), Error> {
if self.epoch != 0 || self.estimated.is_some() {
return Err(Error::InputNotAccepted);
}
@ -94,7 +145,9 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
.or_insert_with(BTreeSet::new)
.insert(input);
// Multicast BVAL
Ok(AgreementMessage::BVal(self.epoch, input))
self.messages
.push_back(AgreementMessage::BVal(self.epoch, input));
Ok(())
}
/// Acceptance check to be performed before setting the input value.
@ -102,33 +155,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
self.epoch == 0 && self.estimated.is_none()
}
/// Receive input from a remote node.
///
/// Outputs an optional agreement result and a queue of agreement messages
/// to remote nodes. There can be up to 2 messages.
pub fn handle_agreement_message(
&mut self,
sender_id: &NodeUid,
message: &AgreementMessage,
) -> Result<AgreementOutput, Error> {
match *message {
// The algorithm instance has already terminated.
_ if self.terminated => Err(Error::Terminated),
AgreementMessage::BVal(epoch, b) if epoch == self.epoch => {
self.handle_bval(sender_id, b)
}
AgreementMessage::Aux(epoch, b) if epoch == self.epoch => self.handle_aux(sender_id, b),
// Epoch does not match. Ignore the message.
_ => Ok((None, VecDeque::new())),
}
}
fn handle_bval(&mut self, sender_id: &NodeUid, b: bool) -> Result<AgreementOutput, Error> {
let mut outgoing = VecDeque::new();
fn handle_bval(&mut self, sender_id: &NodeUid, b: bool) -> Result<(), Error> {
self.received_bval
.entry(sender_id.clone())
.or_insert_with(BTreeSet::new)
@ -148,14 +175,13 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
// where w ∈ bin_values_r
if bin_values_was_empty {
// Send an AUX message at most once per epoch.
outgoing.push_back(AgreementMessage::Aux(self.epoch, b));
self.messages
.push_back(AgreementMessage::Aux(self.epoch, b));
// Receive the AUX message locally.
self.received_aux.insert(self.uid.clone(), b);
}
let (decision, maybe_message) = self.try_coin();
outgoing.extend(maybe_message);
Ok((decision, outgoing))
self.try_coin();
}
// upon receiving BVAL_r(b) messages from f + 1 nodes, if
// BVAL_r(b) has not been sent, multicast BVAL_r(b)
@ -168,23 +194,18 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
.or_insert_with(BTreeSet::new)
.insert(b);
// Multicast BVAL.
outgoing.push_back(AgreementMessage::BVal(self.epoch, b));
Ok((None, outgoing))
} else {
Ok((None, outgoing))
self.messages
.push_back(AgreementMessage::BVal(self.epoch, b));
}
Ok(())
}
fn handle_aux(&mut self, sender_id: &NodeUid, b: bool) -> Result<AgreementOutput, Error> {
fn handle_aux(&mut self, sender_id: &NodeUid, b: bool) -> Result<(), Error> {
self.received_aux.insert(sender_id.clone(), b);
let mut outgoing = VecDeque::new();
if !self.bin_values.is_empty() {
let (decision, maybe_message) = self.try_coin();
outgoing.extend(maybe_message);
Ok((decision, outgoing))
} else {
Ok((None, outgoing))
self.try_coin();
}
Ok(())
}
/// AUX_r messages such that the set of values carried by those messages is
@ -216,11 +237,11 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
/// to compute the next decision estimate and outputs the optional decision
/// value. The function may start the next epoch. In that case, it also
/// returns a message for broadcast.
fn try_coin(&mut self) -> (Option<bool>, Vec<AgreementMessage>) {
fn try_coin(&mut self) {
let (count_aux, vals) = self.count_aux();
if count_aux < self.num_nodes - self.num_faulty_nodes {
// Continue waiting for the (N - f) AUX messages.
return (None, Vec::new());
return;
}
debug!("{:?} try_coin in epoch {}", self.uid, self.epoch);
@ -231,7 +252,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
// Check the termination condition: "continue looping until both a
// value b is output in some round r, and the value Coin_r' = b for
// some round r' > r."
self.terminated = self.terminated || self.output == Some(coin);
self.terminated = self.terminated || self.decision == Some(coin);
if self.terminated {
debug!("Agreement instance {:?} terminated", self.uid);
}
@ -247,29 +268,27 @@ impl<NodeUid: Clone + Debug + Eq + Hash> Agreement<NodeUid> {
self.uid, self.epoch
);
let decision = if vals.len() != 1 {
if vals.len() != 1 {
self.estimated = Some(coin);
None
} else {
// NOTE: `vals` has exactly one element due to `vals.len() == 1`
let v: Vec<bool> = vals.into_iter().collect();
let b = v[0];
self.estimated = Some(b);
// Outputting a value is allowed only once.
if self.output.is_none() && b == coin {
if self.decision.is_none() && b == coin {
// Output the agreement value.
self.output = Some(b);
// Latch the decided state.
self.decision = Some(b);
debug!("Agreement instance {:?} output: {}", self.uid, b);
self.output
} else {
None
}
};
let b = self.estimated.unwrap();
self.sent_bval.insert(b);
let bval_msg = AgreementMessage::BVal(self.epoch, b);
(decision, vec![bval_msg])
self.messages
.push_back(AgreementMessage::BVal(self.epoch, b));
}
}

35
src/common_subset.rs
View File

@ -99,6 +99,10 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> DistAlgorithm for CommonSubset<No
}
fn terminated(&self) -> bool {
debug!(
"Termination check. Terminated Agreement instances: {:?}",
self.agreement_instances.values().all(Agreement::terminated)
);
self.messages.is_empty() && self.agreement_instances.values().all(Agreement::terminated)
}
@ -168,9 +172,12 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
fn on_broadcast_result(&mut self, uid: &NodeUid) -> Result<(), Error> {
if let Some(agreement_instance) = self.agreement_instances.get_mut(&uid) {
if agreement_instance.accepts_input() {
let msg = agreement_instance.set_input(true)?;
self.messages
.push_back(Target::All.message(Message::Agreement(uid.clone(), msg)));
agreement_instance.set_input(true)?;
self.messages.extend(
agreement_instance
.message_iter()
.map(|msg| msg.map(|a_msg| Message::Agreement(uid.clone(), a_msg))),
);
}
} else {
return Err(Error::NoSuchBroadcastInstance);
@ -224,17 +231,16 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
return Ok(());
}
// Send the message to the agreement instance.
let (opt_output, msgs) =
agreement_instance.handle_agreement_message(sender_id, &amessage)?;
agreement_instance.handle_message(sender_id, amessage.clone())?;
self.messages.extend(
msgs.into_iter().map(|a_msg| {
Target::All.message(Message::Agreement(proposer_id.clone(), a_msg))
}),
agreement_instance
.message_iter()
.map(|msg| msg.map(|a_msg| Message::Agreement(proposer_id.clone(), a_msg))),
);
input_result = opt_output;
input_result = agreement_instance.next_output();
} else {
debug!("Proposer {:?} does not exist.", proposer_id);
return Ok(());
return Err(Error::NoSuchAgreementInstance);
}
if let Some(output) = input_result {
@ -267,10 +273,13 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
// input 0 to each instance of BA that has not yet been provided input.
for agreement_instance in self.agreement_instances.values_mut() {
if agreement_instance.accepts_input() {
let msg = agreement_instance.set_input(false)?;
agreement_instance.set_input(false)?;
let uid = agreement_instance.our_id().clone();
self.messages
.push_back(Target::All.message(Message::Agreement(uid, msg)));
self.messages.extend(
agreement_instance
.message_iter()
.map(|msg| msg.map(|a_msg| Message::Agreement(uid.clone(), a_msg))),
);
}
}
Ok(())

101
tests/agreement.rs
View File

@ -21,26 +21,27 @@ extern crate env_logger;
use std::collections::{BTreeMap, VecDeque};
use hbbft::agreement::{Agreement, AgreementMessage};
use hbbft::messaging::{DistAlgorithm, Target, TargetedMessage};
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
struct NodeId(usize);
struct NodeUid(usize);
/// The queue of messages of a particular Agreement instance.
type InstanceQueue = VecDeque<AgreementMessage>;
type MessageQueue = VecDeque<TargetedMessage<AgreementMessage, NodeUid>>;
struct TestNode {
/// Sender ID.
id: NodeId,
id: NodeUid,
/// The only agreement instance.
agreement: Agreement<NodeId>,
agreement: Agreement<NodeUid>,
/// Queue of tuples of a sender ID and a message.
queue: VecDeque<(NodeId, AgreementMessage)>,
queue: VecDeque<(NodeUid, AgreementMessage)>,
/// All outputs
outputs: Vec<bool>,
}
impl TestNode {
fn new(id: NodeId, agreement: Agreement<NodeId>) -> TestNode {
fn new(id: NodeUid, agreement: Agreement<NodeUid>) -> TestNode {
TestNode {
id,
agreement,
@ -49,46 +50,63 @@ impl TestNode {
}
}
fn handle_message(&mut self) -> (Option<bool>, InstanceQueue) {
fn handle_message(&mut self) -> (Option<bool>, MessageQueue) {
let output;
let (sender_id, message) = self.queue
.pop_front()
.expect("popping a message off the queue");
let (output, messages) = self.agreement
.handle_agreement_message(&sender_id, &message)
self.agreement
.handle_message(&sender_id, message)
.expect("handling an agreement message");
debug!("{:?} produced messages: {:?}", self.id, messages);
if let Some(output) = output {
self.outputs.push(output);
if let Some(b) = self.agreement.next_output() {
self.outputs.push(b);
output = Some(b);
} else {
output = None;
}
let messages = self.agreement.message_iter().collect();
debug!("{:?} produced messages: {:?}", self.id, messages);
(output, messages)
}
}
struct TestNetwork {
nodes: BTreeMap<NodeId, TestNode>,
nodes: BTreeMap<NodeUid, TestNode>,
/// The next node to handle a message in its queue.
scheduled_node_id: NodeId,
scheduled_node_id: NodeUid,
}
impl TestNetwork {
fn new(num_nodes: usize) -> TestNetwork {
// Make a node with an Agreement instance associated with the proposer node 0.
let make_node = |id: NodeId| (id, TestNode::new(id, Agreement::new(NodeId(0), num_nodes)));
let make_node =
|id: NodeUid| (id, TestNode::new(id, Agreement::new(NodeUid(0), num_nodes)));
TestNetwork {
nodes: (0..num_nodes).map(NodeId).map(make_node).collect(),
scheduled_node_id: NodeId(0),
nodes: (0..num_nodes).map(NodeUid).map(make_node).collect(),
scheduled_node_id: NodeUid(0),
}
}
fn dispatch_messages(&mut self, sender_id: NodeId, messages: InstanceQueue) {
fn dispatch_messages(&mut self, sender_id: NodeUid, messages: MessageQueue) {
for message in messages {
for (id, node) in &mut self.nodes {
if *id != sender_id {
debug!(
"Dispatching from {:?} to {:?}: {:?}",
sender_id, id, message
);
node.queue.push_back((sender_id, message.clone()));
match message {
TargetedMessage {
target: Target::Node(id),
message,
} => {
let node = self.nodes.get_mut(&id).expect("finding recipient node");
node.queue.push_back((sender_id, message));
}
TargetedMessage {
target: Target::All,
message,
} => {
// Multicast the message to other nodes.
let _: Vec<()> = self.nodes
.iter_mut()
.filter(|(id, _)| **id != sender_id)
.map(|(_, node)| node.queue.push_back((sender_id, message.clone())))
.collect();
}
}
}
@ -96,7 +114,7 @@ impl TestNetwork {
// Gets a node for receiving a message and picks the next node with a
// non-empty message queue in a cyclic order.
fn pick_node(&mut self) -> NodeId {
fn pick_node(&mut self) -> NodeUid {
let id = self.scheduled_node_id;
// Try a node with a higher ID for fairness.
if let Some(next_id) = self.nodes
@ -117,32 +135,33 @@ impl TestNetwork {
id
}
fn step(&mut self) -> (NodeId, Option<bool>) {
fn step(&mut self) -> (NodeUid, Option<bool>) {
let sender_id = self.pick_node();
let (output, messages) = self.nodes.get_mut(&sender_id).unwrap().handle_message();
self.dispatch_messages(sender_id, messages);
(sender_id, output)
}
fn set_input(&mut self, sender_id: NodeId, input: bool) {
let message = self.nodes
.get_mut(&sender_id)
.unwrap()
.agreement
.set_input(input)
.expect("set input");
self.dispatch_messages(sender_id, VecDeque::from(vec![message]));
fn set_input(&mut self, sender_id: NodeUid, input: bool) {
let messages = {
let instance = &mut self.nodes.get_mut(&sender_id).unwrap().agreement;
instance.set_input(input).expect("set input");
instance.message_iter().collect()
};
self.dispatch_messages(sender_id, messages);
}
}
fn test_agreement(mut network: TestNetwork) -> BTreeMap<NodeId, TestNode> {
fn test_agreement(mut network: TestNetwork) -> BTreeMap<NodeUid, TestNode> {
let _ = env_logger::try_init();
// Pick the first node with a non-empty queue.
network.pick_node();
while network.nodes.values().any(|node| node.outputs.is_empty()) {
let (NodeId(id), output) = network.step();
let (NodeUid(id), output) = network.step();
if let Some(value) = output {
debug!("Node {} output {}", id, value);
}
@ -156,10 +175,10 @@ fn test_agreement(mut network: TestNetwork) -> BTreeMap<NodeId, TestNode> {
fn test_agreement_and_validity_with_1_late_node() {
let mut network = TestNetwork::new(4);
network.set_input(NodeId(0), true);
network.set_input(NodeId(1), true);
network.set_input(NodeId(2), true);
network.set_input(NodeId(3), false);
network.set_input(NodeUid(0), true);
network.set_input(NodeUid(1), true);
network.set_input(NodeUid(2), true);
network.set_input(NodeUid(3), false);
let nodes = test_agreement(network);

15
tests/common_subset.rs
View File

@ -160,19 +160,22 @@ fn test_common_subset(mut network: TestNetwork) -> BTreeMap<NodeUid, TestNode> {
}
#[test]
fn test_common_subset_4_nodes_same_proposed_value() {
fn test_common_subset_3_out_of_4_nodes_propose() {
let proposed_value = Vec::from("Fake news");
let all_ids: HashSet<NodeUid> = (0..4).map(NodeUid).collect();
let mut network = TestNetwork::new(&all_ids);
let expected_node_decision: HashMap<NodeUid, ProposedValue> = all_ids
let proposing_ids: HashSet<NodeUid> = (0..3).map(NodeUid).collect();
let expected_node_decision: HashMap<NodeUid, ProposedValue> = proposing_ids
.iter()
.map(|id| (*id, proposed_value.clone()))
.collect();
network.send_proposed_value(NodeUid(0), proposed_value.clone());
network.send_proposed_value(NodeUid(1), proposed_value.clone());
network.send_proposed_value(NodeUid(2), proposed_value.clone());
network.send_proposed_value(NodeUid(3), proposed_value.clone());
// Nodes propose values.
let _: Vec<()> = proposing_ids
.iter()
.map(|id| network.send_proposed_value(*id, proposed_value.clone()))
.collect();
let nodes = test_common_subset(network);