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

Merge pull request #35 from poanetwork/add-error-chain-crate 

Added error-chain error handling.
This commit is contained in:
Vladimir Komendantskiy 2018-05-21 08:25:37 +01:00 committed by GitHub
parent d223668561 1930ea8b7d
commit 4e00894853
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 143 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.toml
View File

@ -7,12 +7,13 @@ authors = ["Vladimir Komendantskiy <komendantsky@gmail.com>"]
bincode = "1.0.0"
derive_deref = "1.0.1"
env_logger = "0.5.10"
error-chain = "0.11.0"
itertools = "0.7"
log = "0.4.1"
merkle = { git = "https://github.com/afck/merkle.rs", branch = "public-proof" }
protobuf = { version = "2.0.0", optional = true }
rand = "0.4.2"
reed-solomon-erasure = "3.0"
reed-solomon-erasure = "3.1.0"
ring = "^0.12"
serde = "1.0.55"
serde_derive = { version = "1.0.55", optional = true }

2
examples/network/commst.rs
View File

@ -85,7 +85,7 @@ impl<'a, P: Message + 'a, M: Into<P> + From<P> + Send + 'a> CommsTask<'a, P, M>
message: message.into(),
}).unwrap();
}
Err(proto_io::Error::ProtobufError(e)) => {
Err(proto_io::Error(proto_io::ErrorKind::Protobuf(e), _)) => {
warn!("Node {} - Protobuf error {}", node_index, e)
}
Err(e) => {

37
src/agreement.rs
View File

@ -8,6 +8,17 @@ use std::mem;
use messaging::{DistAlgorithm, Target, TargetedMessage};
error_chain!{
types {
Error, ErrorKind, ResultExt, AgreementResult;
}
errors {
InputNotAccepted
Terminated
}
}
/// Messages sent during the binary Byzantine agreement stage.
#[cfg_attr(feature = "serialization-serde", derive(Serialize))]
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
@ -74,7 +85,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> DistAlgorithm for Agreement<NodeU
type Message = AgreementMessage;
type Error = Error;
fn input(&mut self, input: Self::Input) -> Result<(), Self::Error> {
fn input(&mut self, input: Self::Input) -> AgreementResult<()> {
self.set_input(input)
}
@ -83,9 +94,9 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> DistAlgorithm for Agreement<NodeU
&mut self,
sender_id: &Self::NodeUid,
message: Self::Message,
) -> Result<(), Self::Error> {
) -> AgreementResult<()> {
if self.terminated {
return Err(Error::Terminated);
return Err(ErrorKind::Terminated.into());
}
if message.epoch() < self.epoch {
return Ok(()); // Message is obsolete: We are already in a later epoch.
@ -146,9 +157,9 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> Agreement<NodeUid> {
}
/// Sets the input value for agreement.
pub fn set_input(&mut self, input: bool) -> Result<(), Error> {
pub fn set_input(&mut self, input: bool) -> AgreementResult<()> {
if self.epoch != 0 || self.estimated.is_some() {
return Err(Error::InputNotAccepted);
return Err(ErrorKind::InputNotAccepted.into());
}
if self.num_nodes == 1 {
self.decision = Some(input);
@ -167,7 +178,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> Agreement<NodeUid> {
self.epoch == 0 && self.estimated.is_none()
}
fn handle_bval(&mut self, sender_id: &NodeUid, b: bool) -> Result<(), Error> {
fn handle_bval(&mut self, sender_id: &NodeUid, b: bool) -> AgreementResult<()> {
self.received_bval
.entry(sender_id.clone())
.or_insert_with(BTreeSet::new)
@ -197,7 +208,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> Agreement<NodeUid> {
Ok(())
}
fn send_bval(&mut self, b: bool) -> Result<(), Error> {
fn send_bval(&mut self, b: bool) -> AgreementResult<()> {
// Record the value `b` as sent.
self.sent_bval.insert(b);
// Multicast BVAL.
@ -208,12 +219,12 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> Agreement<NodeUid> {
self.handle_bval(&our_uid, b)
}
fn handle_aux(&mut self, sender_id: &NodeUid, b: bool) -> Result<(), Error> {
fn handle_aux(&mut self, sender_id: &NodeUid, b: bool) -> AgreementResult<()> {
self.received_aux.insert(sender_id.clone(), b);
self.try_coin()
}
fn send_aux(&mut self, b: bool) -> Result<(), Error> {
fn send_aux(&mut self, b: bool) -> AgreementResult<()> {
// Multicast AUX.
self.messages
.push_back(AgreementMessage::Aux(self.epoch, b));
@ -251,7 +262,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> 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) -> Result<(), Error> {
fn try_coin(&mut self) -> AgreementResult<()> {
if self.bin_values.is_empty() {
return Ok(());
}
@ -311,9 +322,3 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> Agreement<NodeUid> {
Ok(())
}
}
#[derive(Clone, Debug)]
pub enum Error {
Terminated,
InputNotAccepted,
}

71
src/broadcast.rs
View File

@ -9,6 +9,25 @@ use std::iter;
use messaging::{DistAlgorithm, Target, TargetedMessage};
error_chain!{
types {
Error, ErrorKind, ResultExt, BroadcastResult;
}
foreign_links {
ReedSolomon(rse::Error);
}
errors {
InstanceCannotPropose
NotImplemented
ProofConstructionFailed
RootHashMismatch
Threading
UnknownSender
}
}
/// The three kinds of message sent during the reliable broadcast stage of the
/// consensus algorithm.
#[cfg_attr(feature = "serialization-serde", derive(Serialize, Deserialize))]
@ -105,9 +124,9 @@ impl<N: Eq + Debug + Clone + Ord> DistAlgorithm for Broadcast<N> {
type Message = BroadcastMessage;
type Error = Error;
fn input(&mut self, input: Self::Input) -> Result<(), Self::Error> {
fn input(&mut self, input: Self::Input) -> BroadcastResult<()> {
if self.our_id != self.proposer_id {
return Err(Error::InstanceCannotPropose);
return Err(ErrorKind::InstanceCannotPropose.into());
}
// Split the value into chunks/shards, encode them with erasure codes.
// Assemble a Merkle tree from data and parity shards. Take all proofs
@ -119,9 +138,9 @@ impl<N: Eq + Debug + Clone + Ord> DistAlgorithm for Broadcast<N> {
self.handle_value(&our_id, proof)
}
fn handle_message(&mut self, sender_id: &N, message: Self::Message) -> Result<(), Self::Error> {
fn handle_message(&mut self, sender_id: &N, message: Self::Message) -> BroadcastResult<()> {
if !self.all_uids.contains(sender_id) {
return Err(Error::UnknownSender);
return Err(ErrorKind::UnknownSender.into());
}
match message {
BroadcastMessage::Value(p) => self.handle_value(sender_id, p),
@ -150,7 +169,7 @@ impl<N: Eq + Debug + Clone + Ord> DistAlgorithm for Broadcast<N> {
impl<N: Eq + Debug + Clone + Ord> Broadcast<N> {
/// Creates a new broadcast instance to be used by node `our_id` which expects a value proposal
/// from node `proposer_id`.
pub fn new(our_id: N, proposer_id: N, all_uids: BTreeSet<N>) -> Result<Self, Error> {
pub fn new(our_id: N, proposer_id: N, all_uids: BTreeSet<N>) -> BroadcastResult<Self> {
let num_nodes = all_uids.len();
let num_faulty_nodes = (num_nodes - 1) / 3;
let parity_shard_num = 2 * num_faulty_nodes;
@ -180,7 +199,7 @@ impl<N: Eq + Debug + Clone + Ord> Broadcast<N> {
/// scheme. The returned value contains the shard assigned to this
/// node. That shard doesn't need to be sent anywhere. It gets recorded in
/// the broadcast instance.
fn send_shards(&mut self, mut value: Vec<u8>) -> Result<Proof<Vec<u8>>, Error> {
fn send_shards(&mut self, mut value: Vec<u8>) -> BroadcastResult<Proof<Vec<u8>>> {
let data_shard_num = self.coding.data_shard_count();
let parity_shard_num = self.coding.parity_shard_count();
@ -229,14 +248,14 @@ impl<N: Eq + Debug + Clone + Ord> Broadcast<N> {
let mtree = MerkleTree::from_vec(&digest::SHA256, shards_t);
// Default result in case of `gen_proof` error.
let mut result = Err(Error::ProofConstructionFailed);
let mut result = Err(ErrorKind::ProofConstructionFailed.into());
assert_eq!(self.num_nodes, mtree.iter().count());
// Send each proof to a node.
for (leaf_value, uid) in mtree.iter().zip(&self.all_uids) {
let proof = mtree
.gen_proof(leaf_value.to_vec())
.ok_or(Error::ProofConstructionFailed)?;
.ok_or(ErrorKind::ProofConstructionFailed)?;
if *uid == self.our_id {
// The proof is addressed to this node.
result = Ok(proof);
@ -251,7 +270,7 @@ impl<N: Eq + Debug + Clone + Ord> Broadcast<N> {
}
/// Handles a received echo and verifies the proof it contains.
fn handle_value(&mut self, sender_id: &N, p: Proof<Vec<u8>>) -> Result<(), Error> {
fn handle_value(&mut self, sender_id: &N, p: Proof<Vec<u8>>) -> BroadcastResult<()> {
// If the sender is not the proposer, this is not the first `Value` or the proof is invalid,
// ignore.
if *sender_id != self.proposer_id {
@ -279,7 +298,7 @@ impl<N: Eq + Debug + Clone + Ord> Broadcast<N> {
}
/// Handles a received `Echo` message.
fn handle_echo(&mut self, sender_id: &N, p: Proof<Vec<u8>>) -> Result<(), Error> {
fn handle_echo(&mut self, sender_id: &N, p: Proof<Vec<u8>>) -> BroadcastResult<()> {
// If the proof is invalid or the sender has already sent `Echo`, ignore.
if self.echos.contains_key(sender_id) {
info!(
@ -310,7 +329,7 @@ impl<N: Eq + Debug + Clone + Ord> Broadcast<N> {
}
/// Handles a received `Ready` message.
fn handle_ready(&mut self, sender_id: &N, hash: &[u8]) -> Result<(), Error> {
fn handle_ready(&mut self, sender_id: &N, hash: &[u8]) -> BroadcastResult<()> {
// If the sender has already sent a `Ready` before, ignore.
if self.readys.contains_key(sender_id) {
info!(
@ -335,7 +354,7 @@ impl<N: Eq + Debug + Clone + Ord> Broadcast<N> {
/// Checks whether the condition for output are met for this hash, and if so, sets the output
/// value.
fn compute_output(&mut self, hash: &[u8]) -> Result<(), Error> {
fn compute_output(&mut self, hash: &[u8]) -> BroadcastResult<()> {
if self.decided || self.count_readys(hash) <= 2 * self.num_faulty_nodes
|| self.count_echos(hash) <= self.num_faulty_nodes
{
@ -417,7 +436,7 @@ enum Coding {
impl Coding {
/// Creates a new `Coding` instance with the given number of shards.
fn new(data_shard_num: usize, parity_shard_num: usize) -> Result<Self, Error> {
fn new(data_shard_num: usize, parity_shard_num: usize) -> BroadcastResult<Self> {
Ok(if parity_shard_num > 0 {
let rs = ReedSolomon::new(data_shard_num, parity_shard_num)?;
Coding::ReedSolomon(Box::new(rs))
@ -443,7 +462,7 @@ impl Coding {
}
/// Constructs (and overwrites) the parity shards.
fn encode(&self, slices: &mut [&mut [u8]]) -> Result<(), Error> {
fn encode(&self, slices: &mut [&mut [u8]]) -> BroadcastResult<()> {
match *self {
Coding::ReedSolomon(ref rs) => rs.encode(slices)?,
Coding::Trivial(_) => (),
@ -452,7 +471,7 @@ impl Coding {
}
/// If enough shards are present, reconstructs the missing ones.
fn reconstruct_shards(&self, shards: &mut [Option<Box<[u8]>>]) -> Result<(), Error> {
fn reconstruct_shards(&self, shards: &mut [Option<Box<[u8]>>]) -> BroadcastResult<()> {
match *self {
Coding::ReedSolomon(ref rs) => rs.reconstruct_shards(shards)?,
Coding::Trivial(_) => {
@ -465,30 +484,12 @@ impl Coding {
}
}
/// Errors returned by the broadcast instance.
#[derive(Debug, Clone)]
pub enum Error {
RootHashMismatch,
Threading,
ProofConstructionFailed,
ReedSolomon(rse::Error),
InstanceCannotPropose,
NotImplemented,
UnknownSender,
}
impl From<rse::Error> for Error {
fn from(err: rse::Error) -> Error {
Error::ReedSolomon(err)
}
}
fn decode_from_shards<T>(
leaf_values: &mut [Option<Box<[u8]>>],
coding: &Coding,
data_shard_num: usize,
root_hash: &[u8],
) -> Result<T, Error>
) -> BroadcastResult<T>
where
T: From<Vec<u8>>,
{
@ -513,7 +514,7 @@ where
// NOTE: The paper does not define the meaning of *abort*. But it is
// sensible not to continue trying to reconstruct the tree after this
// point. This instance must have received incorrect shards.
Err(Error::RootHashMismatch)
Err(ErrorKind::RootHashMismatch.into())
} else {
// Reconstruct the value from the data shards.
Ok(glue_shards(mtree, data_shard_num))

66
src/common_subset.rs
View File

@ -14,6 +14,25 @@ use broadcast::{Broadcast, BroadcastMessage};
use fmt::HexBytes;
use messaging::{DistAlgorithm, Target, TargetedMessage};
error_chain!{
types {
Error, ErrorKind, ResultExt, CommonSubsetResult;
}
links {
Agreement(agreement::Error, agreement::ErrorKind);
Broadcast(broadcast::Error, broadcast::ErrorKind);
}
errors {
MultipleAgreementResults
NoSuchAgreementInstance
NoSuchBroadcastInstance
NotImplemented
UnexpectedMessage
}
}
// TODO: Make this a generic argument of `Broadcast`.
type ProposedValue = Vec<u8>;
// Type of output from the Common Subset message handler.
@ -98,7 +117,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> DistAlgorithm for CommonSubset<No
type Message = Message<NodeUid>;
type Error = Error;
fn input(&mut self, input: Self::Input) -> Result<(), Self::Error> {
fn input(&mut self, input: Self::Input) -> CommonSubsetResult<()> {
self.send_proposed_value(input)
}
@ -106,7 +125,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> DistAlgorithm for CommonSubset<No
&mut self,
sender_id: &Self::NodeUid,
message: Self::Message,
) -> Result<(), Self::Error> {
) -> CommonSubsetResult<()> {
match message {
Message::Broadcast(p_id, b_msg) => self.handle_broadcast(sender_id, &p_id, b_msg),
Message::Agreement(p_id, a_msg) => self.handle_agreement(sender_id, &p_id, a_msg),
@ -131,7 +150,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> DistAlgorithm for CommonSubset<No
}
impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
pub fn new(uid: NodeUid, all_uids: &BTreeSet<NodeUid>) -> Result<Self, Error> {
pub fn new(uid: NodeUid, all_uids: &BTreeSet<NodeUid>) -> CommonSubsetResult<Self> {
let num_nodes = all_uids.len();
let num_faulty_nodes = (num_nodes - 1) / 3;
@ -170,7 +189,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
/// Common Subset input message handler. It receives a value for broadcast
/// and redirects it to the corresponding broadcast instance.
pub fn send_proposed_value(&mut self, value: ProposedValue) -> Result<(), Error> {
pub fn send_proposed_value(&mut self, value: ProposedValue) -> CommonSubsetResult<()> {
let uid = self.uid.clone();
// Upon receiving input v_i , input v_i to RBC_i. See Figure 2.
self.process_broadcast(&uid, |bc| bc.input(value))
@ -183,7 +202,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
sender_id: &NodeUid,
proposer_id: &NodeUid,
bmessage: BroadcastMessage,
) -> Result<(), Error> {
) -> CommonSubsetResult<()> {
self.process_broadcast(proposer_id, |bc| bc.handle_message(sender_id, bmessage))
}
@ -194,7 +213,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
sender_id: &NodeUid,
proposer_id: &NodeUid,
amessage: AgreementMessage,
) -> Result<(), Error> {
) -> CommonSubsetResult<()> {
// Send the message to the local instance of Agreement
self.process_agreement(proposer_id, |agreement| {
agreement.handle_message(sender_id, amessage)
@ -203,14 +222,14 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
/// Upon delivery of v_j from RBC_j, if input has not yet been provided to
/// BA_j, then provide input 1 to BA_j. See Figure 11.
fn process_broadcast<F>(&mut self, proposer_id: &NodeUid, f: F) -> Result<(), Error>
fn process_broadcast<F>(&mut self, proposer_id: &NodeUid, f: F) -> CommonSubsetResult<()>
where
F: FnOnce(&mut Broadcast<NodeUid>) -> Result<(), broadcast::Error>,
{
let value = {
let broadcast = self.broadcast_instances
.get_mut(proposer_id)
.ok_or(Error::NoSuchBroadcastInstance)?;
.ok_or(ErrorKind::NoSuchBroadcastInstance)?;
f(broadcast)?;
self.messages.extend_broadcast(&proposer_id, broadcast);
if let Some(output) = broadcast.next_output() {
@ -231,14 +250,14 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
/// Callback to be invoked on receipt of the decision value of the Agreement
/// instance `uid`.
fn process_agreement<F>(&mut self, proposer_id: &NodeUid, f: F) -> Result<(), Error>
fn process_agreement<F>(&mut self, proposer_id: &NodeUid, f: F) -> CommonSubsetResult<()>
where
F: FnOnce(&mut Agreement<NodeUid>) -> Result<(), agreement::Error>,
{
let value = {
let agreement = self.agreement_instances
.get_mut(proposer_id)
.ok_or(Error::NoSuchAgreementInstance)?;
.ok_or(ErrorKind::NoSuchAgreementInstance)?;
if agreement.terminated() {
return Ok(());
}
@ -254,7 +273,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
.insert(proposer_id.clone(), value)
.is_some()
{
return Err(Error::MultipleAgreementResults);
return Err(ErrorKind::MultipleAgreementResults.into());
}
debug!(
"{:?} Updated Agreement results: {:?}",
@ -270,7 +289,7 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
self.messages.extend_agreement(uid, agreement);
if let Some(output) = agreement.next_output() {
if self.agreement_results.insert(uid.clone(), output).is_some() {
return Err(Error::MultipleAgreementResults);
return Err(ErrorKind::MultipleAgreementResults.into());
}
}
}
@ -321,26 +340,3 @@ impl<NodeUid: Clone + Debug + Eq + Hash + Ord> CommonSubset<NodeUid> {
}
}
}
#[derive(Clone, Debug)]
pub enum Error {
UnexpectedMessage,
NotImplemented,
NoSuchBroadcastInstance,
NoSuchAgreementInstance,
MultipleAgreementResults,
Broadcast(broadcast::Error),
Agreement(agreement::Error),
}
impl From<broadcast::Error> for Error {
fn from(err: broadcast::Error) -> Error {
Error::Broadcast(err)
}
}
impl From<agreement::Error> for Error {
fn from(err: agreement::Error) -> Error {
Error::Agreement(err)
}
}

58
src/honey_badger.rs
View File

@ -12,6 +12,25 @@ use serde::Serialize;
use common_subset::{self, CommonSubset};
use messaging::{DistAlgorithm, TargetedMessage};
error_chain!{
types {
Error, ErrorKind, ResultExt, HoneyBadgerResult;
}
links {
CommonSubset(common_subset::Error, common_subset::ErrorKind);
}
foreign_links {
Bincode(Box<bincode::ErrorKind>);
}
errors {
OwnIdMissing
UnknownSender
}
}
/// An instance of the Honey Badger Byzantine fault tolerant consensus algorithm.
pub struct HoneyBadger<T, N: Eq + Hash + Ord + Clone> {
/// The buffer of transactions that have not yet been included in any output batch.
@ -46,13 +65,13 @@ where
type Message = Message<N>;
type Error = Error;
fn input(&mut self, input: Self::Input) -> Result<(), Self::Error> {
fn input(&mut self, input: Self::Input) -> HoneyBadgerResult<()> {
self.add_transactions(iter::once(input))
}
fn handle_message(&mut self, sender_id: &N, message: Self::Message) -> Result<(), Self::Error> {
fn handle_message(&mut self, sender_id: &N, message: Self::Message) -> HoneyBadgerResult<()> {
if !self.all_uids.contains(sender_id) {
return Err(Error::UnknownSender);
return Err(ErrorKind::UnknownSender.into());
}
match message {
Message::CommonSubset(epoch, cs_msg) => {
@ -85,14 +104,14 @@ where
N: Eq + Hash + Ord + Clone + Debug,
{
/// Returns a new Honey Badger instance with the given parameters, starting at epoch `0`.
pub fn new<I, TI>(id: N, all_uids_iter: I, batch_size: usize, txs: TI) -> Result<Self, Error>
pub fn new<I, TI>(id: N, all_uids_iter: I, batch_size: usize, txs: TI) -> HoneyBadgerResult<Self>
where
I: IntoIterator<Item = N>,
TI: IntoIterator<Item = T>,
{
let all_uids: BTreeSet<N> = all_uids_iter.into_iter().collect();
if !all_uids.contains(&id) {
return Err(Error::OwnIdMissing);
return Err(ErrorKind::OwnIdMissing.into());
}
let mut honey_badger = HoneyBadger {
buffer: txs.into_iter().collect(),
@ -109,13 +128,13 @@ where
}
/// Adds transactions into the buffer.
pub fn add_transactions<I: IntoIterator<Item = T>>(&mut self, txs: I) -> Result<(), Error> {
pub fn add_transactions<I: IntoIterator<Item = T>>(&mut self, txs: I) -> HoneyBadgerResult<()> {
self.buffer.extend(txs);
Ok(())
}
/// Proposes a new batch in the current epoch.
fn propose(&mut self) -> Result<(), Error> {
fn propose(&mut self) -> HoneyBadgerResult<()> {
let proposal = self.choose_transactions()?;
let cs = match self.common_subsets.entry(self.epoch) {
Entry::Occupied(entry) => entry.into_mut(),
@ -130,7 +149,7 @@ where
/// Returns a random choice of `batch_size / all_uids.len()` buffered transactions, and
/// serializes them.
fn choose_transactions(&self) -> Result<Vec<u8>, Error> {
fn choose_transactions(&self) -> HoneyBadgerResult<Vec<u8>> {
let mut rng = rand::thread_rng();
let amount = cmp::max(1, self.batch_size / self.all_uids.len());
let batch_size = cmp::min(self.batch_size, self.buffer.len());
@ -151,7 +170,7 @@ where
sender_id: &N,
epoch: u64,
message: common_subset::Message<N>,
) -> Result<(), Error> {
) -> HoneyBadgerResult<()> {
{
// Borrow the instance for `epoch`, or create it.
let cs = match self.common_subsets.entry(epoch) {
@ -259,24 +278,3 @@ impl<N: Clone + Debug + Eq + Hash + Ord> MessageQueue<N> {
self.extend(cs.message_iter().map(convert));
}
}
/// A Honey Badger error.
#[derive(Debug)]
pub enum Error {
OwnIdMissing,
UnknownSender,
CommonSubset(common_subset::Error),
Bincode(Box<bincode::ErrorKind>),
}
impl From<common_subset::Error> for Error {
fn from(err: common_subset::Error) -> Error {
Error::CommonSubset(err)
}
}
impl From<Box<bincode::ErrorKind>> for Error {
fn from(err: Box<bincode::ErrorKind>) -> Error {
Error::Bincode(err)
}
}

2
src/lib.rs
View File

@ -9,6 +9,8 @@ extern crate bincode;
#[macro_use(Deref, DerefMut)]
extern crate derive_deref;
#[macro_use]
extern crate error_chain;
#[macro_use]
extern crate log;
extern crate itertools;
extern crate merkle;

49
src/proto_io.rs
View File

@ -1,6 +1,6 @@
//! Protobuf message IO task structure.
use protobuf::{self, Message};
use protobuf::{self, Message, ProtobufError};
use std::io::{Read, Write};
use std::marker::PhantomData;
use std::net::TcpStream;
@ -12,31 +12,26 @@ use std::{cmp, io};
/// TODO: Replace it with a proper handshake at connection initiation.
const FRAME_START: u32 = 0x2C0F_FEE5;
#[derive(Debug)]
pub enum Error {
IoError(io::Error),
EncodeError,
DecodeError,
FrameStartMismatch,
// ProtocolError,
ProtobufError(protobuf::ProtobufError),
}
error_chain!{
types {
Error, ErrorKind, ResultExt, ProtoIoResult;
}
impl From<io::Error> for Error {
fn from(err: io::Error) -> Error {
Error::IoError(err)
foreign_links {
Io(io::Error);
Protobuf(ProtobufError);
}
errors {
Decode
Encode
FrameStartMismatch
}
}
impl From<protobuf::ProtobufError> for Error {
fn from(err: protobuf::ProtobufError) -> Error {
Error::ProtobufError(err)
}
}
fn encode_u32_to_be(value: u32, buffer: &mut [u8]) -> Result<(), Error> {
fn encode_u32_to_be(value: u32, buffer: &mut [u8]) -> ProtoIoResult<()> {
if buffer.len() < 4 {
return Err(Error::EncodeError);
return Err(ErrorKind::Encode.into());
}
let value = value.to_le();
buffer[0] = ((value & 0xFF00_0000) >> 24) as u8;
@ -46,9 +41,9 @@ fn encode_u32_to_be(value: u32, buffer: &mut [u8]) -> Result<(), Error> {
Ok(())
}
fn decode_u32_from_be(buffer: &[u8]) -> Result<u32, Error> {
fn decode_u32_from_be(buffer: &[u8]) -> ProtoIoResult<u32> {
if buffer.len() < 4 {
return Err(Error::DecodeError);
return Err(ErrorKind::Decode.into());
}
let mut result = u32::from(buffer[0]);
result <<= 8;
@ -88,11 +83,11 @@ impl<S: Read + Write, M: Message> ProtoIo<S, M>
}
}
pub fn recv(&mut self) -> Result<M, Error> {
pub fn recv(&mut self) -> ProtoIoResult<M> {
self.stream.read_exact(&mut self.buffer[0..4])?;
let frame_start = decode_u32_from_be(&self.buffer[0..4])?;
if frame_start != FRAME_START {
return Err(Error::FrameStartMismatch);
return Err(ErrorKind::FrameStartMismatch.into());
};
self.stream.read_exact(&mut self.buffer[0..4])?;
let size = decode_u32_from_be(&self.buffer[0..4])? as usize;
@ -106,10 +101,10 @@ impl<S: Read + Write, M: Message> ProtoIo<S, M>
message_v.extend_from_slice(slice);
}
protobuf::parse_from_bytes(&message_v).map_err(Error::ProtobufError)
protobuf::parse_from_bytes(&message_v).map_err(|e| e.into())
}
pub fn send(&mut self, message: &M) -> Result<(), Error> {
pub fn send(&mut self, message: &M) -> ProtoIoResult<()> {
let mut buffer: [u8; 4] = [0; 4];
// Wrap stream
let mut stream = protobuf::CodedOutputStream::new(&mut self.stream);