Skip to content
This repository has been archived by the owner on Apr 17, 2019. It is now read-only.

Commit

Permalink
BFT OS components (#1703)
Browse files Browse the repository at this point in the history 
* BFT OS trunk

* On demand OS transport (#1635)

* On demand OS connection manager (#1645)

* On-demand ordering gate and related fixes (#1675)

Signed-off-by: Andrei Lebedev <[email protected]>
  • Loading branch information
@lebdron @nickaleks
lebdron authored and nickaleks committed Sep 10, 2018
1 parent 6dc9ac9 commit 3945302
Show file tree
Hide file tree
Showing 96 changed files with 2,963 additions and 875 deletions.
10 changes: 8 additions & 2 deletions cmake/functions.cmake
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,10 +82,16 @@ endfunction()
function(compile_proto_only_grpc_to_cpp PROTO)
string(REGEX REPLACE "\\.proto$" ".grpc.pb.h" GEN_GRPC_PB_HEADER ${PROTO})
string(REGEX REPLACE "\\.proto$" ".grpc.pb.cc" GEN_GRPC_PB ${PROTO})
if(TESTING)
# Generate gRPC mock classes for services
set(GENERATE_MOCKS "generate_mock_code=true:")
string(REGEX REPLACE "\\.proto$" "_mock.grpc.pb.h" GEN_GRPC_PB_MOCK_HEADER ${PROTO})
set(TEST_OUTPUT ${SCHEMA_OUT_DIR}/${GEN_GRPC_PB_MOCK_HEADER})
endif(TESTING)
add_custom_command(
OUTPUT ${SCHEMA_OUT_DIR}/${GEN_GRPC_PB_HEADER} ${SCHEMA_OUT_DIR}/${GEN_GRPC_PB}
OUTPUT ${SCHEMA_OUT_DIR}/${GEN_GRPC_PB_HEADER} ${SCHEMA_OUT_DIR}/${GEN_GRPC_PB} ${TEST_OUTPUT}
COMMAND ${CMAKE_COMMAND} -E env LD_LIBRARY_PATH=${protobuf_LIBRARY_DIR}:$ENV{LD_LIBRARY_PATH} "${protoc_EXECUTABLE}"
ARGS -I${protobuf_INCLUDE_DIR} -I${CMAKE_CURRENT_SOURCE_DIR} ${ARGN} --grpc_out=${SCHEMA_OUT_DIR} --plugin=protoc-gen-grpc="${grpc_CPP_PLUGIN}" ${PROTO}
ARGS -I${protobuf_INCLUDE_DIR} -I${CMAKE_CURRENT_SOURCE_DIR} ${ARGN} --grpc_out=${GENERATE_MOCKS}${SCHEMA_OUT_DIR} --plugin=protoc-gen-grpc="${grpc_CPP_PLUGIN}" ${PROTO}
DEPENDS grpc_cpp_plugin ${SCHEMA_PATH}/${PROTO}
WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
)
Expand Down
2 changes: 1 addition & 1 deletion iroha-cli/interactive/impl/interactive_common_cli.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ namespace iroha_cli {
params_description.value().end(),
[&params](auto &param) {
using namespace iroha;
promptString(param) | [&](auto &val) {
promptString(param) | [&](auto &&val) {
if (not val.empty()) {
// Update input cache
param.cache = val;
Expand Down
10 changes: 6 additions & 4 deletions irohad/consensus/yac/cluster_order.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,11 @@
#ifndef IROHA_CLUSTER_ORDER_HPP
#define IROHA_CLUSTER_ORDER_HPP

#include <boost/optional.hpp>
#include <vector>
#include <memory>
#include <vector>

#include <boost/optional.hpp>
#include "consensus/yac/yac_types.hpp"

namespace shared_model {
namespace interface {
Expand Down Expand Up @@ -65,7 +67,7 @@ namespace iroha {
std::vector<std::shared_ptr<shared_model::interface::Peer>> getPeers()
const;

size_t getNumberOfPeers() const;
PeersNumberType getNumberOfPeers() const;

virtual ~ClusterOrdering() = default;

Expand All @@ -77,7 +79,7 @@ namespace iroha {
std::vector<std::shared_ptr<shared_model::interface::Peer>> order);

std::vector<std::shared_ptr<shared_model::interface::Peer>> order_;
uint32_t index_ = 0;
PeersNumberType index_ = 0;
};
} // namespace yac
} // namespace consensus
Expand Down
10 changes: 5 additions & 5 deletions irohad/consensus/yac/impl/supermajority_checker_impl.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@ namespace iroha {
and peersSubset(signatures, peers);
}

bool SupermajorityCheckerImpl::checkSize(uint64_t current,
uint64_t all) const {
bool SupermajorityCheckerImpl::checkSize(PeersNumberType current,
PeersNumberType all) const {
if (current > all) {
return false;
}
Expand All @@ -55,9 +55,9 @@ namespace iroha {
}));
}

bool SupermajorityCheckerImpl::hasReject(uint64_t frequent,
uint64_t voted,
uint64_t all) const {
bool SupermajorityCheckerImpl::hasReject(PeersNumberType frequent,
PeersNumberType voted,
PeersNumberType all) const {
auto not_voted = all - voted;
return not checkSize(frequent + not_voted, all);
}
Expand Down
9 changes: 5 additions & 4 deletions irohad/consensus/yac/impl/supermajority_checker_impl.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,17 +40,18 @@ namespace iroha {
const std::vector<std::shared_ptr<shared_model::interface::Peer>>
&peers) const override;

virtual bool checkSize(uint64_t current, uint64_t all) const override;
virtual bool checkSize(PeersNumberType current,
PeersNumberType all) const override;

virtual bool peersSubset(
const shared_model::interface::types::SignatureRangeType
&signatures,
const std::vector<std::shared_ptr<shared_model::interface::Peer>>
&peers) const override;

virtual bool hasReject(uint64_t frequent,
uint64_t voted,
uint64_t all) const override;
virtual bool hasReject(PeersNumberType frequent,
PeersNumberType voted,
PeersNumberType all) const override;
};
} // namespace yac
} // namespace consensus
Expand Down
207 changes: 62 additions & 145 deletions irohad/consensus/yac/impl/yac.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,11 +46,6 @@ namespace iroha {
return result;
}

template <typename T>
static std::string cryptoError(const std::initializer_list<T> &votes) {
return cryptoError<std::initializer_list<T>>(votes);
}

std::shared_ptr<Yac> Yac::create(
YacVoteStorage vote_storage,
std::shared_ptr<YacNetwork> network,
Expand Down Expand Up @@ -83,41 +78,23 @@ namespace iroha {

cluster_order_ = order;
auto vote = crypto_->getVote(hash);
// TODO 10.06.2018 andrei: IR-1407 move YAC propagation strategy to a
// separate entity
votingStep(vote);
}

rxcpp::observable<CommitMessage> Yac::on_commit() {
rxcpp::observable<Answer> Yac::onOutcome() {
return notifier_.get_observable();
}

// ------|Network notifications|------

void Yac::on_vote(VoteMessage vote) {
std::lock_guard<std::mutex> guard(mutex_);
if (crypto_->verify(vote)) {
applyVote(findPeer(vote), vote);
} else {
log_->warn(cryptoError({vote}));
}
}

void Yac::on_commit(CommitMessage commit) {
void Yac::onState(std::vector<VoteMessage> state) {
std::lock_guard<std::mutex> guard(mutex_);
if (crypto_->verify(commit)) {
// Commit does not contain data about peer which sent the message
applyCommit(boost::none, commit);
if (crypto_->verify(state)) {
applyState(state);
} else {
log_->warn(cryptoError(commit.votes));
}
}

void Yac::on_reject(RejectMessage reject) {
std::lock_guard<std::mutex> guard(mutex_);
if (crypto_->verify(reject)) {
// Reject does not contain data about peer which sent the message
applyReject(boost::none, reject);
} else {
log_->warn(cryptoError(reject.votes));
log_->warn(cryptoError(state));
}
}

Expand All @@ -133,7 +110,7 @@ namespace iroha {
vote.hash.proposal_hash,
vote.hash.block_hash);

network_->send_vote(cluster_order_.currentLeader(), vote);
network_->sendState(cluster_order_.currentLeader(), {vote});
cluster_order_.switchToNext();
if (cluster_order_.hasNext()) {
timer_->invokeAfterDelay([this, vote] { this->votingStep(vote); });
Expand All @@ -157,137 +134,77 @@ namespace iroha {

// ------|Apply data|------

const char *kRejectMsg = "reject case";
const char *kRejectOnHashMsg = "Reject case on hash {} achieved";

void Yac::applyCommit(
boost::optional<std::shared_ptr<shared_model::interface::Peer>> from,
const CommitMessage &commit) {
void Yac::applyState(const std::vector<VoteMessage> &state) {
auto answer =
vote_storage_.store(commit, cluster_order_.getNumberOfPeers());
answer | [&](const auto &answer) {
auto proposal_hash = getProposalHash(commit.votes).value();
auto already_processed =
vote_storage_.getProcessingState(proposal_hash);
if (not already_processed) {
vote_storage_.markAsProcessedState(proposal_hash);
visit_in_place(answer,
[&](const CommitMessage &commit) {
notifier_.get_subscriber().on_next(commit);
},
[&](const RejectMessage &reject) {
log_->warn(kRejectMsg);
// TODO 14/08/17 Muratov: work on reject case
// IR-497
});
}
this->closeRound();
};
}
vote_storage_.store(state, cluster_order_.getNumberOfPeers());

void Yac::applyReject(
boost::optional<std::shared_ptr<shared_model::interface::Peer>> from,
const RejectMessage &reject) {
auto answer =
vote_storage_.store(reject, cluster_order_.getNumberOfPeers());
answer | [&](const auto &answer) {
auto proposal_hash = getProposalHash(reject.votes).value();
auto already_processed =
vote_storage_.getProcessingState(proposal_hash);
// TODO 10.06.2018 andrei: IR-1407 move YAC propagation strategy to a
// separate entity

if (not already_processed) {
vote_storage_.markAsProcessedState(proposal_hash);
visit_in_place(answer,
[&](const RejectMessage &reject) {
log_->warn(kRejectMsg);
// TODO 14/08/17 Muratov: work on reject case
// IR-497
},
[&](const CommitMessage &commit) {
this->propagateCommit(commit);
notifier_.get_subscriber().on_next(commit);
});
answer | [&](const auto &answer) {
auto &proposal_hash = state.at(0).hash.proposal_hash;

/*
* It is possible that a new peer with an outdated peers list may
* collect an outcome from a smaller number of peers which are
* included in set of `f` peers in the system. The new peer will not
* accept our message with valid supermajority because he cannot apply
* votes from unknown peers.
*/
if (state.size() > 1) {
// some peer has already collected commit/reject, so it is sent
if (vote_storage_.getProcessingState(proposal_hash)
== ProposalState::kNotSentNotProcessed) {
vote_storage_.nextProcessingState(proposal_hash);
log_->info(
"Received supermajority of votes for {}, skip propagation",
proposal_hash);
}
}
this->closeRound();
};
}

void Yac::applyVote(
boost::optional<std::shared_ptr<shared_model::interface::Peer>> from,
const VoteMessage &vote) {
if (from) {
log_->info("Apply vote: {} from ledger peer {}",
vote.hash.block_hash,
(*from)->address());
} else {
log_->info("Apply vote: {} from unknown peer {}",
vote.hash.block_hash,
vote.signature->publicKey().hex());
}

auto answer =
vote_storage_.store(vote, cluster_order_.getNumberOfPeers());

answer | [&](const auto &answer) {
auto &proposal_hash = vote.hash.proposal_hash;
auto already_processed =
auto processing_state =
vote_storage_.getProcessingState(proposal_hash);

if (not already_processed) {
vote_storage_.markAsProcessedState(proposal_hash);
visit_in_place(answer,
[&](const CommitMessage &commit) {
// propagate for all
log_->info("Propagate commit {} to whole network",
vote.hash.block_hash);
this->propagateCommit(commit);
notifier_.get_subscriber().on_next(commit);
},
[&](const RejectMessage &reject) {
// propagate reject for all
log_->info(kRejectOnHashMsg, proposal_hash);
this->propagateReject(reject);
});
} else {
from | [&](const auto &from) {
visit_in_place(answer,
[&](const CommitMessage &commit) {
log_->info("Propagate commit {} directly to {}",
vote.hash.block_hash,
from->address());
this->propagateCommitDirectly(*from, commit);
},
[&](const RejectMessage &reject) {
log_->info(kRejectOnHashMsg, proposal_hash);
this->propagateRejectDirectly(*from, reject);
});
};
auto votes = [](const auto &state) { return state.votes; };

switch (processing_state) {
case ProposalState::kNotSentNotProcessed:
vote_storage_.nextProcessingState(proposal_hash);
log_->info("Propagate state {} to whole network", proposal_hash);
this->propagateState(visit_in_place(answer, votes));
break;
case ProposalState::kSentNotProcessed:
vote_storage_.nextProcessingState(proposal_hash);
log_->info("Pass outcome for {} to pipeline", proposal_hash);
this->closeRound();
notifier_.get_subscriber().on_next(answer);
break;
case ProposalState::kSentProcessed:
if (state.size() == 1) {
this->findPeer(state.at(0)) | [&](const auto &from) {
log_->info("Propagate state {} directly to {}",
proposal_hash,
from->address());
this->propagateStateDirectly(*from,
visit_in_place(answer, votes));
};
}
break;
}
};
}

// ------|Propagation|------

void Yac::propagateCommit(const CommitMessage &msg) {
for (const auto &peer : cluster_order_.getPeers()) {
propagateCommitDirectly(*peer, msg);
}
}

void Yac::propagateCommitDirectly(const shared_model::interface::Peer &to,
const CommitMessage &msg) {
network_->send_commit(to, msg);
}

void Yac::propagateReject(const RejectMessage &msg) {
void Yac::propagateState(const std::vector<VoteMessage> &msg) {
for (const auto &peer : cluster_order_.getPeers()) {
propagateRejectDirectly(*peer, msg);
propagateStateDirectly(*peer, msg);
}
}

void Yac::propagateRejectDirectly(const shared_model::interface::Peer &to,
const RejectMessage &msg) {
network_->send_reject(std::move(to), std::move(msg));
void Yac::propagateStateDirectly(const shared_model::interface::Peer &to,
const std::vector<VoteMessage> &msg) {
network_->sendState(to, msg);
}

} // namespace yac
Expand Down
Loading

0 comments on commit 3945302

Please sign in to comment.