aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/client.cpp4
-rw-r--r--src/consensus.cpp2
-rw-r--r--src/hotstuff.cpp45
-rw-r--r--src/hotstuff_app.cpp39
-rw-r--r--src/hotstuff_client.cpp123
5 files changed, 155 insertions, 58 deletions
diff --git a/src/client.cpp b/src/client.cpp
index 482c23d..486594a 100644
--- a/src/client.cpp
+++ b/src/client.cpp
@@ -11,9 +11,9 @@ void MsgClient::gen_reqcmd(const Command &cmd) {
set_payload(std::move(s));
}
-void MsgClient::parse_reqcmd(CommandDummy &cmd) const {
+void MsgClient::parse_reqcmd(command_t &cmd, HotStuffCore *hsc) const {
DataStream s(get_payload());
- s >> cmd;
+ cmd = hsc->parse_cmd(s);
}
void MsgClient::gen_respcmd(const uint256_t &cmd_hash, const Finality &fin) {
diff --git a/src/consensus.cpp b/src/consensus.cpp
index 7749558..e42fb49 100644
--- a/src/consensus.cpp
+++ b/src/consensus.cpp
@@ -186,7 +186,7 @@ void HotStuffCore::on_receive_proposal(const Proposal &prop) {
(opinion ?
create_part_cert(*priv_key, bnew->get_hash()) :
nullptr),
- nullptr));
+ this));
}
void HotStuffCore::on_receive_vote(const Vote &vote) {
diff --git a/src/hotstuff.cpp b/src/hotstuff.cpp
index 9582531..f4454d4 100644
--- a/src/hotstuff.cpp
+++ b/src/hotstuff.cpp
@@ -70,6 +70,26 @@ void MsgHotStuff::parse_rfetchblk(std::vector<block_t> &blks, HotStuffCore *hsc)
}
}
+ReplicaID HotStuffBase::add_command(command_t cmd) {
+ ReplicaID proposer = pmaker->get_proposer();
+ if (proposer != get_id())
+ return proposer;
+ cmd_pending.push(storage->add_cmd(cmd));
+ if (cmd_pending.size() >= blk_size)
+ {
+ std::vector<command_t> cmds;
+ for (uint32_t i = 0; i < blk_size; i++)
+ {
+ cmds.push_back(cmd_pending.front());
+ cmd_pending.pop();
+ }
+ pmaker->beat().then([this, cmds = std::move(cmds)]() {
+ on_propose(cmds);
+ });
+ }
+ return proposer;
+}
+
void HotStuffBase::add_replica(ReplicaID idx, const NetAddr &addr,
pubkey_bt &&pub_key) {
HotStuffCore::add_replica(idx, addr, std::move(pub_key));
@@ -387,8 +407,6 @@ HotStuffBase::HotStuffBase(uint32_t blk_size,
part_delivery_time_min(double_inf),
part_delivery_time_max(0)
{
- if (pmaker == nullptr)
- this->pmaker = new PaceMakerDummy(this);
/* register the handlers for msg from replicas */
pn.reg_handler(PROPOSE, std::bind(&HotStuffBase::propose_handler, this, _1, _2));
pn.reg_handler(VOTE, std::bind(&HotStuffBase::vote_handler, this, _1, _2));
@@ -405,11 +423,16 @@ void HotStuffBase::do_broadcast_proposal(const Proposal &prop) {
}
void HotStuffBase::do_vote(ReplicaID last_proposer, const Vote &vote) {
- MsgHotStuff vote_msg;
- vote_msg.gen_vote(vote);
pmaker->next_proposer(last_proposer)
- .then([this, vote_msg](ReplicaID proposer) {
- pn.send_msg(vote_msg, get_config().get_addr(proposer));
+ .then([this, vote](ReplicaID proposer) {
+ if (proposer == get_id())
+ on_receive_vote(vote);
+ else
+ {
+ MsgHotStuff vote_msg;
+ vote_msg.gen_vote(vote);
+ pn.send_msg(vote_msg, get_config().get_addr(proposer));
+ }
});
}
@@ -422,6 +445,15 @@ void HotStuffBase::do_decide(const command_t &cmd) {
}
}
+void HotStuffBase::do_forward(const uint256_t &cmd_hash, ReplicaID rid) {
+ auto it = decision_waiting.find(cmd_hash);
+ if (it != decision_waiting.end())
+ {
+ it->second.reject(rid);
+ decision_waiting.erase(it);
+ }
+}
+
HotStuffBase::~HotStuffBase() {}
void HotStuffBase::start(bool eb_loop) {
@@ -429,6 +461,7 @@ void HotStuffBase::start(bool eb_loop) {
uint32_t nfaulty = pn.all_peers().size() / 3;
if (nfaulty == 0)
LOG_WARN("too few replicas in the system to tolerate any failure");
+ pmaker->init(this);
on_init(nfaulty);
if (eb_loop)
eb.dispatch();
diff --git a/src/hotstuff_app.cpp b/src/hotstuff_app.cpp
index e0c9c3c..5f21fec 100644
--- a/src/hotstuff_app.cpp
+++ b/src/hotstuff_app.cpp
@@ -70,6 +70,13 @@ class HotStuffApp: public HotStuff {
/** checks if a cmd is decided */
inline void client_check_cmd_handler(const MsgClient &, conn_client_t);
+ Finality get_finality(const command_t &cmd) const {
+ hotstuff::block_t blk = cmd->get_container();
+ return Finality(get_id(),
+ cmd->get_decision(),
+ blk ? blk->get_hash() : uint256_t());
+ }
+
/** The callback function to print stat */
inline void print_stat_cb(evutil_socket_t, short);
@@ -203,7 +210,7 @@ HotStuffApp::HotStuffApp(uint32_t blk_size,
NetAddr clisten_addr,
const EventContext &eb):
HotStuff(blk_size, parent_limit, idx, raw_privkey,
- plisten_addr, eb),
+ plisten_addr, eb, new hotstuff::PaceMakerDummyFixed(1)),
stat_period(stat_period),
eb(eb),
cn(eb),
@@ -217,9 +224,9 @@ HotStuffApp::HotStuffApp(uint32_t blk_size,
void HotStuffApp::client_request_cmd_handler(const MsgClient &msg, conn_client_t conn_) {
auto conn = static_pointer_cast<ClientNetwork<MsgClient>::Conn>(conn_);
const NetAddr addr = conn->get_addr();
- command_t cmd = new CommandDummy();
+ command_t cmd;
std::vector<promise_t> pms;
- msg.parse_reqcmd(static_cast<CommandDummy &>(*cmd));
+ msg.parse_reqcmd(cmd, this);
bool flag = true;
#ifndef HOTSTUFF_DISABLE_TX_VERIFY
@@ -229,20 +236,30 @@ void HotStuffApp::client_request_cmd_handler(const MsgClient &msg, conn_client_t
{
LOG_WARN("invalid client cmd");
MsgClient resp;
- resp.gen_respcmd(cmd->get_hash(), Finality(-1, uint256_t()));
+ resp.gen_respcmd(cmd->get_hash(), Finality(get_id(), -1, uint256_t()));
cn.send_msg(resp, addr);
}
else
{
const uint256_t cmd_hash = cmd->get_hash();
- add_command(cmd);
- /** wait for the decision of tx */
- LOG_DEBUG("processing client cmd %.10s", get_hex(cmd_hash).c_str());
- async_decide(cmd_hash).then([this, addr](command_t cmd) {
+ ReplicaID rid = add_command(cmd);
+ if (rid == get_id())
+ {
+ /** wait for the decision of tx */
+ LOG_DEBUG("processing client cmd %.10s", get_hex(cmd_hash).c_str());
+ async_decide(cmd_hash).then([this, addr](command_t cmd) {
+ MsgClient resp;
+ resp.gen_respcmd(cmd->get_hash(), get_finality(cmd));
+ cn.send_msg(resp, addr);
+ });
+ }
+ else
+ {
+ LOG_INFO("redirect");
MsgClient resp;
- resp.gen_respcmd(cmd->get_hash(), cmd->get_finality());
+ resp.gen_respcmd(cmd_hash, Finality(rid, 0, cmd_hash));
cn.send_msg(resp, addr);
- });
+ }
}
}
@@ -254,7 +271,7 @@ void HotStuffApp::client_check_cmd_handler(const MsgClient &msg, conn_client_t c
MsgClient resp;
command_t cmd = storage->find_cmd(cmd_hash);
Finality fin;
- if (cmd) fin = cmd->get_finality();
+ if (cmd) fin = get_finality(cmd);
resp.gen_respcmd(cmd_hash, fin);
cn.send_msg(resp, addr);
}
diff --git a/src/hotstuff_client.cpp b/src/hotstuff_client.cpp
index 8ee90d7..9478d36 100644
--- a/src/hotstuff_client.cpp
+++ b/src/hotstuff_client.cpp
@@ -13,6 +13,7 @@ using salticidae::ElapsedTime;
using salticidae::trim_all;
using salticidae::split;
+using hotstuff::ReplicaID;
using hotstuff::NetAddr;
using hotstuff::EventContext;
using hotstuff::Event;
@@ -20,18 +21,13 @@ using hotstuff::uint256_t;
using hotstuff::bytearray_t;
using hotstuff::MsgClient;
using hotstuff::CommandDummy;
+using hotstuff::command_t;
using hotstuff::Finality;
+EventContext eb;
size_t max_async_num = 10;
int max_iter_num = 100;
-
-struct Request {
- ElapsedTime et;
- Request() { et.start(); }
-};
-
-std::unordered_map<int, salticidae::RingBuffer> buffers;
-std::unordered_map<const uint256_t, Request> waiting;
+ReplicaID proposer;
int connect(const NetAddr &node) {
int fd;
@@ -48,6 +44,54 @@ int connect(const NetAddr &node) {
return fd;
}
+void on_receive(int);
+
+struct Request {
+ ReplicaID rid;
+ command_t cmd;
+ ElapsedTime et;
+ Request(ReplicaID rid, const command_t &cmd):
+ rid(rid), cmd(cmd) { et.start(); }
+};
+
+struct Conn {
+ int fd;
+ Event on_receive_ev;
+
+ Conn(const NetAddr &addr):
+ fd(connect(addr)),
+ on_receive_ev(eb, fd, EV_READ, [this](int fd, short) {
+ on_receive(fd);
+ on_receive_ev.add();
+ }) { on_receive_ev.add(); }
+
+ Conn(Conn &&other):
+ fd(other.fd),
+ on_receive_ev(eb, fd, EV_READ, [this](int fd, short) {
+ on_receive(fd);
+ on_receive_ev.add();
+ }) {
+ other.fd = -1;
+ other.on_receive_ev.del();
+ on_receive_ev.add();
+ }
+
+ ~Conn() { if (fd != -1) close(fd); }
+};
+
+std::unordered_map<int, salticidae::RingBuffer> buffers;
+std::unordered_map<const uint256_t, Request> waiting;
+std::unordered_map<ReplicaID, Conn> conns;
+std::vector<NetAddr> replicas;
+
+
+void setup(ReplicaID rid) {
+ proposer = rid;
+ auto it = conns.find(rid);
+ if (it == conns.end())
+ conns.insert(std::make_pair(rid, Conn(replicas[rid])));
+}
+
void write_msg(int fd, const MsgClient &msg) {
bytearray_t msg_data = msg.serialize();
if (write(fd, msg_data.data(), msg_data.size()) != (ssize_t)msg_data.size())
@@ -86,17 +130,17 @@ void read_msg(int fd, MsgClient &msg) {
}
}
-void try_send(int fd) {
+void try_send() {
while (waiting.size() < max_async_num && max_iter_num)
{
auto cmd = CommandDummy::make_cmd();
MsgClient msg;
msg.gen_reqcmd(*cmd);
- write_msg(fd, msg);
+ write_msg(conns.find(proposer)->second.fd, msg);
HOTSTUFF_LOG_INFO("send new cmd %.10s",
get_hex(cmd->get_hash()).c_str());
waiting.insert(std::make_pair(
- cmd->get_hash(), Request()));
+ cmd->get_hash(), Request(proposer, cmd)));
if (max_iter_num > 0)
max_iter_num--;
}
@@ -111,15 +155,31 @@ void on_receive(int fd) {
if (!msg.verify_checksum())
HOTSTUFF_LOG_ERROR("incorrect checksum %08x", msg.get_checksum());
msg.parse_respcmd(cmd_hash, fin);
+ auto it = waiting.find(cmd_hash);
+ if (fin.rid != proposer)
+ {
+ HOTSTUFF_LOG_INFO("reconnect to the new proposer");
+ setup(fin.rid);
+ }
+ if (fin.rid != it->second.rid)
+ {
+ MsgClient msg;
+ msg.gen_reqcmd(*(waiting.find(cmd_hash)->second.cmd));
+ write_msg(conns.find(proposer)->second.fd, msg);
+ HOTSTUFF_LOG_INFO("resend cmd %.10s",
+ get_hex(cmd_hash).c_str());
+ it->second.et.start();
+ it->second.rid = proposer;
+ return;
+ }
HOTSTUFF_LOG_INFO(
"fd %d got response for %.10s: <decision=%d, blk=%.10s>",
fd, get_hex(cmd_hash).c_str(),
fin.decision,
get_hex(fin.blk_hash).c_str());
- auto it = waiting.find(cmd_hash);
if (it == waiting.end()) return;
waiting.erase(it);
- try_send(fd);
+ try_send();
}
std::pair<std::string, std::string> split_ip_port_cport(const std::string &s) {
@@ -127,53 +187,40 @@ std::pair<std::string, std::string> split_ip_port_cport(const std::string &s) {
return std::make_pair(ret[0], ret[1]);
}
-Event *on_receive_ev;
int main(int argc, char **argv) {
Config config("hotstuff.conf");
- std::vector<NetAddr> peers2;
- EventContext eb;
- auto opt_idx = Config::OptValInt::create(-1);
- auto opt_server_addr = Config::OptValStr::create("127.0.0.1:2234");
+ auto opt_idx = Config::OptValInt::create(0);
auto opt_replicas = Config::OptValStrVec::create();
auto opt_max_iter_num = Config::OptValInt::create();
try {
config.add_opt("idx", opt_idx, Config::SET_VAL);
- config.add_opt("server", opt_server_addr, Config::SET_VAL);
config.add_opt("replica", opt_replicas, Config::APPEND);
config.add_opt("ntx", opt_max_iter_num, Config::SET_VAL);
config.parse(argc, argv);
auto idx = opt_idx->get();
max_iter_num = opt_max_iter_num->get();
- std::vector<std::pair<std::string, std::string>> replicas;
+ std::vector<std::pair<std::string, std::string>> raw;
for (const auto &s: opt_replicas->get())
{
auto res = trim_all(split(s, ","));
assert(res.size() == 2);
- replicas.push_back(std::make_pair(res[0], res[1]));
+ raw.push_back(std::make_pair(res[0], res[1]));
}
- NetAddr server(opt_server_addr->get());
- if (-1 < idx && (size_t)idx < replicas.size() &&
- replicas.size() > 0)
+ if (!(0 <= idx && (size_t)idx < raw.size() &&
+ raw.size() > 0))
+ throw std::invalid_argument("out of range");
+ for (const auto &p: raw)
{
- for (const auto &p: replicas)
- {
- auto _p = split_ip_port_cport(p.first);
- size_t _;
- peers2.push_back(NetAddr(NetAddr(_p.first).ip, htons(stoi(_p.second, &_))));
- }
- server = peers2[idx];
+ auto _p = split_ip_port_cport(p.first);
+ size_t _;
+ replicas.push_back(NetAddr(NetAddr(_p.first).ip, htons(stoi(_p.second, &_))));
}
- int fd = connect(server);
- on_receive_ev = new Event{eb, fd, EV_READ, [](int fd, short) {
- on_receive(fd);
- on_receive_ev->add();
- }};
- on_receive_ev->add();
- try_send(fd);
+ setup(idx);
+ try_send();
eb.dispatch();
} catch (hotstuff::HotStuffError &e) {
HOTSTUFF_LOG_ERROR("exception: %s", std::string(e).c_str());