aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/hotstuff.cpp59
-rw-r--r--src/hotstuff_app.cpp158
-rw-r--r--src/hotstuff_client.cpp116
3 files changed, 156 insertions, 177 deletions
diff --git a/src/hotstuff.cpp b/src/hotstuff.cpp
index 1912946..a8cc625 100644
--- a/src/hotstuff.cpp
+++ b/src/hotstuff.cpp
@@ -107,7 +107,10 @@ void HotStuffBase::add_replica(ReplicaID idx, const NetAddr &addr,
pubkey_bt &&pub_key) {
HotStuffCore::add_replica(idx, addr, std::move(pub_key));
if (addr != listen_addr)
+ {
+ peers.insert(addr);
pn.add_peer(addr);
+ }
}
void HotStuffBase::on_fetch_blk(const block_t &blk) {
@@ -221,8 +224,8 @@ promise_t HotStuffBase::async_deliver_blk(const uint256_t &blk_hash,
return static_cast<promise_t &>(pm);
}
-void HotStuffBase::propose_handler(MsgPropose &&msg, Conn &conn) {
- const NetAddr &peer = conn.get_peer();
+void HotStuffBase::propose_handler(MsgPropose &&msg, const Net::conn_t &conn) {
+ const NetAddr &peer = conn->get_peer();
msg.postponed_parse(this);
auto &prop = msg.proposal;
block_t blk = prop.blk;
@@ -235,8 +238,8 @@ void HotStuffBase::propose_handler(MsgPropose &&msg, Conn &conn) {
});
}
-void HotStuffBase::vote_handler(MsgVote &&msg, Conn &conn) {
- const NetAddr &peer = conn.get_peer();
+void HotStuffBase::vote_handler(MsgVote &&msg, const Net::conn_t &conn) {
+ const NetAddr &peer = conn->get_peer();
msg.postponed_parse(this);
//auto &vote = msg.vote;
RcObj<Vote> v(new Vote(std::move(msg.vote)));
@@ -255,8 +258,8 @@ void HotStuffBase::vote_handler(MsgVote &&msg, Conn &conn) {
});
}
-void HotStuffBase::req_blk_handler(MsgReqBlock &&msg, Conn &conn) {
- const NetAddr replica = conn.get_peer();
+void HotStuffBase::req_blk_handler(MsgReqBlock &&msg, const Net::conn_t &conn) {
+ const NetAddr replica = conn->get_peer();
auto &blk_hashes = msg.blk_hashes;
std::vector<promise_t> pms;
for (const auto &h: blk_hashes)
@@ -272,7 +275,7 @@ void HotStuffBase::req_blk_handler(MsgReqBlock &&msg, Conn &conn) {
});
}
-void HotStuffBase::resp_blk_handler(MsgRespBlock &&msg, Conn &) {
+void HotStuffBase::resp_blk_handler(MsgRespBlock &&msg, const Net::conn_t &) {
msg.postponed_parse(this);
for (const auto &blk: msg.blks)
if (blk) on_fetch_blk(blk);
@@ -310,28 +313,10 @@ void HotStuffBase::print_stat() const {
part_delivery_time_min = double_inf;
part_delivery_time_max = 0;
#ifdef HOTSTUFF_MSG_STAT
- LOG_INFO("-- sent opcode (10s) --");
- auto &sent_op = pn.get_sent_by_opcode();
- for (auto &op: sent_op)
- {
- auto &val = op.second;
- LOG_INFO("%02x: %lu, %.2fBpm", op.first,
- val.first, val.first ? val.second / double(val.first) : 0);
- val.first = val.second = 0;
- }
- LOG_INFO("-- recv opcode (10s) --");
- auto &recv_op = pn.get_recv_by_opcode();
- for (auto &op: recv_op)
- {
- auto &val = op.second;
- LOG_INFO("%02x: %lu, %.2fBpm", op.first,
- val.first, val.first ? val.second / double(val.first) : 0);
- val.first = val.second = 0;
- }
LOG_INFO("--- replica msg. (10s) ---");
size_t _nsent = 0;
size_t _nrecv = 0;
- for (const auto &replica: pn.all_peers())
+ for (const auto &replica: peers)
{
auto conn = pn.get_peer_conn(replica);
if (conn == nullptr) continue;
@@ -361,14 +346,15 @@ HotStuffBase::HotStuffBase(uint32_t blk_size,
privkey_bt &&priv_key,
NetAddr listen_addr,
pacemaker_bt pmaker,
- EventContext eb,
- size_t nworker):
+ EventContext ec,
+ size_t nworker,
+ const Net::Config &config):
HotStuffCore(rid, std::move(priv_key)),
listen_addr(listen_addr),
blk_size(blk_size),
- eb(eb),
- vpool(eb, nworker),
- pn(eb),
+ ec(ec),
+ vpool(ec, nworker),
+ pn(ec, config),
pmaker(std::move(pmaker)),
fetched(0), delivered(0),
@@ -387,12 +373,13 @@ HotStuffBase::HotStuffBase(uint32_t blk_size,
pn.reg_handler(salticidae::generic_bind(&HotStuffBase::vote_handler, this, _1, _2));
pn.reg_handler(salticidae::generic_bind(&HotStuffBase::req_blk_handler, this, _1, _2));
pn.reg_handler(salticidae::generic_bind(&HotStuffBase::resp_blk_handler, this, _1, _2));
+ pn.start();
pn.listen(listen_addr);
}
void HotStuffBase::do_broadcast_proposal(const Proposal &prop) {
MsgPropose prop_msg(prop);
- for (const auto &replica: pn.all_peers())
+ for (const auto &replica: peers)
pn.send_msg(prop_msg, replica);
}
@@ -422,15 +409,15 @@ void HotStuffBase::do_decide(Finality &&fin) {
HotStuffBase::~HotStuffBase() {}
-void HotStuffBase::start(bool eb_loop) {
+void HotStuffBase::start(bool ec_loop) {
/* ((n - 1) + 1 - 1) / 3 */
- uint32_t nfaulty = pn.all_peers().size() / 3;
+ uint32_t nfaulty = peers.size() / 3;
if (nfaulty == 0)
LOG_WARN("too few replicas in the system to tolerate any failure");
on_init(nfaulty);
pmaker->init(this);
- if (eb_loop)
- eb.dispatch();
+ if (ec_loop)
+ ec.dispatch();
}
}
diff --git a/src/hotstuff_app.cpp b/src/hotstuff_app.cpp
index 014fe16..ee57a3a 100644
--- a/src/hotstuff_app.cpp
+++ b/src/hotstuff_app.cpp
@@ -29,7 +29,7 @@ using salticidae::static_pointer_cast;
using salticidae::trim_all;
using salticidae::split;
-using hotstuff::Event;
+using hotstuff::TimerEvent;
using hotstuff::EventContext;
using hotstuff::NetAddr;
using hotstuff::HotStuffError;
@@ -55,17 +55,17 @@ class HotStuffApp: public HotStuff {
/** Network messaging between a replica and its client. */
ClientNetwork<opcode_t> cn;
/** Timer object to schedule a periodic printing of system statistics */
- Event ev_stat_timer;
+ TimerEvent ev_stat_timer;
/** Timer object to monitor the progress for simple impeachment */
- Event impeach_timer;
+ TimerEvent impeach_timer;
/** The listen address for client RPC */
NetAddr clisten_addr;
std::unordered_map<const uint256_t, promise_t> unconfirmed;
- using Conn = ClientNetwork<opcode_t>::Conn;
+ using conn_t = ClientNetwork<opcode_t>::conn_t;
- void client_request_cmd_handler(MsgReqCmd &&, Conn &);
+ void client_request_cmd_handler(MsgReqCmd &&, const conn_t &);
static command_t parse_cmd(DataStream &s) {
auto cmd = new CommandDummy();
@@ -75,7 +75,7 @@ class HotStuffApp: public HotStuff {
void reset_imp_timer() {
impeach_timer.del();
- impeach_timer.add_with_timeout(impeach_timeout);
+ impeach_timer.add(impeach_timeout);
}
void state_machine_execute(const Finality &fin) override {
@@ -113,10 +113,6 @@ std::pair<std::string, std::string> split_ip_port_cport(const std::string &s) {
return std::make_pair(ret[0], ret[1]);
}
-void signal_handler(int) {
- throw HotStuffError("got terminal signal");
-}
-
salticidae::BoxObj<HotStuffApp> papp = nullptr;
int main(int argc, char **argv) {
@@ -125,9 +121,6 @@ int main(int argc, char **argv) {
ElapsedTime elapsed;
elapsed.start();
- signal(SIGTERM, signal_handler);
- signal(SIGINT, signal_handler);
-
auto opt_blk_size = Config::OptValInt::create(1);
auto opt_parent_limit = Config::OptValInt::create(-1);
auto opt_stat_period = Config::OptValDouble::create(10);
@@ -157,74 +150,72 @@ int main(int argc, char **argv) {
config.add_opt("help", opt_help, Config::SWITCH_ON, 'h', "show this help info");
EventContext ec;
-#ifdef HOTSTUFF_NORMAL_LOG
- try {
-#endif
- config.parse(argc, argv);
- if (opt_help->get())
- {
- config.print_help();
- exit(0);
- }
- auto idx = opt_idx->get();
- auto client_port = opt_client_port->get();
- std::vector<std::pair<std::string, std::string>> replicas;
- for (const auto &s: opt_replicas->get())
- {
- auto res = trim_all(split(s, ","));
- if (res.size() != 2)
- throw HotStuffError("invalid replica info");
- replicas.push_back(std::make_pair(res[0], res[1]));
- }
+ config.parse(argc, argv);
+ if (opt_help->get())
+ {
+ config.print_help();
+ exit(0);
+ }
+ auto idx = opt_idx->get();
+ auto client_port = opt_client_port->get();
+ std::vector<std::pair<std::string, std::string>> replicas;
+ for (const auto &s: opt_replicas->get())
+ {
+ auto res = trim_all(split(s, ","));
+ if (res.size() != 2)
+ throw HotStuffError("invalid replica info");
+ replicas.push_back(std::make_pair(res[0], res[1]));
+ }
- if (!(0 <= idx && (size_t)idx < replicas.size()))
- throw HotStuffError("replica idx out of range");
- std::string binding_addr = replicas[idx].first;
- if (client_port == -1)
- {
- auto p = split_ip_port_cport(binding_addr);
- size_t idx;
- try {
- client_port = stoi(p.second, &idx);
- } catch (std::invalid_argument &) {
- throw HotStuffError("client port not specified");
- }
+ if (!(0 <= idx && (size_t)idx < replicas.size()))
+ throw HotStuffError("replica idx out of range");
+ std::string binding_addr = replicas[idx].first;
+ if (client_port == -1)
+ {
+ auto p = split_ip_port_cport(binding_addr);
+ size_t idx;
+ try {
+ client_port = stoi(p.second, &idx);
+ } catch (std::invalid_argument &) {
+ throw HotStuffError("client port not specified");
}
+ }
- NetAddr plisten_addr{split_ip_port_cport(binding_addr).first};
+ NetAddr plisten_addr{split_ip_port_cport(binding_addr).first};
- auto parent_limit = opt_parent_limit->get();
- hotstuff::pacemaker_bt pmaker;
- if (opt_pace_maker->get() == "sticky")
- pmaker = new hotstuff::PaceMakerSticky(parent_limit, opt_qc_timeout->get(), ec);
- else if (opt_pace_maker->get() == "rr")
- pmaker = new hotstuff::PaceMakerRR(parent_limit, opt_qc_timeout->get(), ec);
- else
- pmaker = new hotstuff::PaceMakerDummyFixed(opt_fixed_proposer->get(), parent_limit);
+ auto parent_limit = opt_parent_limit->get();
+ hotstuff::pacemaker_bt pmaker;
+ if (opt_pace_maker->get() == "sticky")
+ pmaker = new hotstuff::PaceMakerSticky(parent_limit, opt_qc_timeout->get(), ec);
+ else if (opt_pace_maker->get() == "rr")
+ pmaker = new hotstuff::PaceMakerRR(parent_limit, opt_qc_timeout->get(), ec);
+ else
+ pmaker = new hotstuff::PaceMakerDummyFixed(opt_fixed_proposer->get(), parent_limit);
- papp = new HotStuffApp(opt_blk_size->get(),
- opt_stat_period->get(),
- opt_imp_timeout->get(),
- idx,
- hotstuff::from_hex(opt_privkey->get()),
- plisten_addr,
- NetAddr("0.0.0.0", client_port),
- std::move(pmaker),
- ec,
- opt_nworker->get());
- for (size_t i = 0; i < replicas.size(); i++)
- {
- auto p = split_ip_port_cport(replicas[i].first);
- papp->add_replica(i, NetAddr(p.first),
- hotstuff::from_hex(replicas[i].second));
- }
- papp->start();
-#ifdef HOTSTUFF_NORMAL_LOG
- } catch (std::exception &e) {
- HOTSTUFF_LOG_INFO("exception: %s", e.what());
- elapsed.stop(true);
+ papp = new HotStuffApp(opt_blk_size->get(),
+ opt_stat_period->get(),
+ opt_imp_timeout->get(),
+ idx,
+ hotstuff::from_hex(opt_privkey->get()),
+ plisten_addr,
+ NetAddr("0.0.0.0", client_port),
+ std::move(pmaker),
+ ec,
+ opt_nworker->get());
+ for (size_t i = 0; i < replicas.size(); i++)
+ {
+ auto p = split_ip_port_cport(replicas[i].first);
+ papp->add_replica(i, NetAddr(p.first),
+ hotstuff::from_hex(replicas[i].second));
}
-#endif
+ auto shutdown = [&](int) { ec.stop(); };
+ salticidae::SigEvent ev_sigint(ec, shutdown);
+ salticidae::SigEvent ev_sigterm(ec, shutdown);
+ ev_sigint.add(SIGINT);
+ ev_sigterm.add(SIGTERM);
+
+ papp->start();
+ elapsed.stop(true);
return 0;
}
@@ -243,15 +234,16 @@ HotStuffApp::HotStuffApp(uint32_t blk_size,
stat_period(stat_period),
impeach_timeout(impeach_timeout),
ec(ec),
- cn(ec),
+ cn(ec, ClientNetwork<opcode_t>::Config()),
clisten_addr(clisten_addr) {
/* register the handlers for msg from clients */
cn.reg_handler(salticidae::generic_bind(&HotStuffApp::client_request_cmd_handler, this, _1, _2));
+ cn.start();
cn.listen(clisten_addr);
}
-void HotStuffApp::client_request_cmd_handler(MsgReqCmd &&msg, Conn &conn) {
- const NetAddr addr = conn.get_addr();
+void HotStuffApp::client_request_cmd_handler(MsgReqCmd &&msg, const conn_t &conn) {
+ const NetAddr addr = conn->get_addr();
auto cmd = parse_cmd(msg.serialized);
const auto &cmd_hash = cmd->get_hash();
std::vector<promise_t> pms;
@@ -275,17 +267,17 @@ void HotStuffApp::client_request_cmd_handler(MsgReqCmd &&msg, Conn &conn) {
}
void HotStuffApp::start() {
- ev_stat_timer = Event(ec, -1, 0, [this](int, short) {
+ ev_stat_timer = TimerEvent(ec, [this](TimerEvent &) {
HotStuff::print_stat();
//HotStuffCore::prune(100);
- ev_stat_timer.add_with_timeout(stat_period);
+ ev_stat_timer.add(stat_period);
});
- ev_stat_timer.add_with_timeout(stat_period);
- impeach_timer = Event(ec, -1, 0, [this](int, short) {
+ ev_stat_timer.add(stat_period);
+ impeach_timer = TimerEvent(ec, [this](TimerEvent &) {
get_pace_maker().impeach();
reset_imp_timer();
});
- impeach_timer.add_with_timeout(impeach_timeout);
+ impeach_timer.add(impeach_timeout);
HOTSTUFF_LOG_INFO("** starting the system with parameters **");
HOTSTUFF_LOG_INFO("blk_size = %lu", blk_size);
HOTSTUFF_LOG_INFO("conns = %lu", HotStuff::size());
diff --git a/src/hotstuff_client.cpp b/src/hotstuff_client.cpp
index d8a6087..e8d7b9e 100644
--- a/src/hotstuff_client.cpp
+++ b/src/hotstuff_client.cpp
@@ -1,6 +1,8 @@
#include <cassert>
#include <random>
#include <signal.h>
+#include <sys/time.h>
+
#include "salticidae/type.h"
#include "salticidae/netaddr.h"
#include "salticidae/network.h"
@@ -11,7 +13,6 @@
#include "hotstuff/client.h"
using salticidae::Config;
-using salticidae::MsgNetwork;
using hotstuff::ReplicaID;
using hotstuff::NetAddr;
@@ -25,7 +26,7 @@ using hotstuff::uint256_t;
using hotstuff::opcode_t;
using hotstuff::command_t;
-EventContext eb;
+EventContext ec;
ReplicaID proposer;
size_t max_async_num;
int max_iter_num;
@@ -42,11 +43,13 @@ struct Request {
rid(rid), cmd(cmd), confirmed(0) { et.start(); }
};
-std::unordered_map<ReplicaID, MsgNetwork<opcode_t>::conn_t> conns;
+using Net = salticidae::MsgNetwork<opcode_t>;
+
+std::unordered_map<ReplicaID, Net::conn_t> conns;
std::unordered_map<const uint256_t, Request> waiting;
std::vector<NetAddr> replicas;
std::vector<std::pair<struct timeval, double>> elapsed;
-MsgNetwork<opcode_t> mn(eb, 10, 10, 4096);
+Net mn(ec, Net::Config());
void connect_all() {
for (size_t i = 0; i < replicas.size(); i++)
@@ -66,7 +69,7 @@ void try_send() {
auto cmd = new CommandDummy(cid, cnt++);
//mn.send_msg(MsgReqCmd(*cmd), *conns.at(proposer));
MsgReqCmd msg(*cmd);
- for (auto &p: conns) mn.send_msg(msg, *(p.second));
+ for (auto &p: conns) mn.send_msg(msg, p.second);
#ifndef HOTSTUFF_ENABLE_BENCHMARK
HOTSTUFF_LOG_INFO("send new cmd %.10s",
get_hex(cmd->get_hash()).c_str());
@@ -78,7 +81,7 @@ void try_send() {
}
}
-void client_resp_cmd_handler(MsgRespCmd &&msg, MsgNetwork<opcode_t>::Conn &) {
+void client_resp_cmd_handler(MsgRespCmd &&msg, const Net::conn_t &) {
auto &fin = msg.fin;
HOTSTUFF_LOG_DEBUG("got %s", std::string(msg.fin).c_str());
const uint256_t &cmd_hash = fin.cmd_hash;
@@ -122,70 +125,67 @@ std::pair<std::string, std::string> split_ip_port_cport(const std::string &s) {
return std::make_pair(ret[0], ret[1]);
}
-void signal_handler(int) {
- throw HotStuffError("got terminal signal");
-}
-
int main(int argc, char **argv) {
Config config("hotstuff.conf");
- signal(SIGTERM, signal_handler);
- signal(SIGINT, signal_handler);
-
auto opt_idx = Config::OptValInt::create(0);
auto opt_replicas = Config::OptValStrVec::create();
auto opt_max_iter_num = Config::OptValInt::create(100);
auto opt_max_async_num = Config::OptValInt::create(10);
auto opt_cid = Config::OptValInt::create(-1);
+ auto shutdown = [&](int) { ec.stop(); };
+ salticidae::SigEvent ev_sigint(ec, shutdown);
+ salticidae::SigEvent ev_sigterm(ec, shutdown);
+ ev_sigint.add(SIGINT);
+ ev_sigterm.add(SIGTERM);
+
mn.reg_handler(client_resp_cmd_handler);
+ mn.start();
+
+ config.add_opt("idx", opt_idx, Config::SET_VAL);
+ config.add_opt("cid", opt_cid, Config::SET_VAL);
+ config.add_opt("replica", opt_replicas, Config::APPEND);
+ config.add_opt("iter", opt_max_iter_num, Config::SET_VAL);
+ config.add_opt("max-async", opt_max_async_num, Config::SET_VAL);
+ config.parse(argc, argv);
+ auto idx = opt_idx->get();
+ max_iter_num = opt_max_iter_num->get();
+ max_async_num = opt_max_async_num->get();
+ std::vector<std::pair<std::string, std::string>> raw;
+ for (const auto &s: opt_replicas->get())
+ {
+ auto res = salticidae::trim_all(salticidae::split(s, ","));
+ if (res.size() != 2)
+ throw HotStuffError("format error");
+ raw.push_back(std::make_pair(res[0], res[1]));
+ }
+
+ if (!(0 <= idx && (size_t)idx < raw.size() && raw.size() > 0))
+ throw std::invalid_argument("out of range");
+ cid = opt_cid->get() != -1 ? opt_cid->get() : idx;
+ for (const auto &p: raw)
+ {
+ auto _p = split_ip_port_cport(p.first);
+ size_t _;
+ replicas.push_back(NetAddr(NetAddr(_p.first).ip, htons(stoi(_p.second, &_))));
+ }
+
+ nfaulty = (replicas.size() - 1) / 3;
+ HOTSTUFF_LOG_INFO("nfaulty = %zu", nfaulty);
+ connect_all();
+ set_proposer(idx);
+ try_send();
+ ec.dispatch();
- try {
- config.add_opt("idx", opt_idx, Config::SET_VAL);
- config.add_opt("cid", opt_cid, Config::SET_VAL);
- config.add_opt("replica", opt_replicas, Config::APPEND);
- config.add_opt("iter", opt_max_iter_num, Config::SET_VAL);
- config.add_opt("max-async", opt_max_async_num, Config::SET_VAL);
- config.parse(argc, argv);
- auto idx = opt_idx->get();
- max_iter_num = opt_max_iter_num->get();
- max_async_num = opt_max_async_num->get();
- std::vector<std::pair<std::string, std::string>> raw;
- for (const auto &s: opt_replicas->get())
- {
- auto res = salticidae::trim_all(salticidae::split(s, ","));
- if (res.size() != 2)
- throw HotStuffError("format error");
- raw.push_back(std::make_pair(res[0], res[1]));
- }
-
- if (!(0 <= idx && (size_t)idx < raw.size() && raw.size() > 0))
- throw std::invalid_argument("out of range");
- cid = opt_cid->get() != -1 ? opt_cid->get() : idx;
- for (const auto &p: raw)
- {
- auto _p = split_ip_port_cport(p.first);
- size_t _;
- replicas.push_back(NetAddr(NetAddr(_p.first).ip, htons(stoi(_p.second, &_))));
- }
-
- nfaulty = (replicas.size() - 1) / 3;
- HOTSTUFF_LOG_INFO("nfaulty = %zu", nfaulty);
- connect_all();
- set_proposer(idx);
- try_send();
- eb.dispatch();
- } catch (HotStuffError &e) {
- HOTSTUFF_LOG_ERROR("exception: %s", std::string(e).c_str());
#ifdef HOTSTUFF_ENABLE_BENCHMARK
- for (const auto &e: elapsed)
- {
- char fmt[64];
- struct tm *tmp = localtime(&e.first.tv_sec);
- strftime(fmt, sizeof fmt, "%Y-%m-%d %H:%M:%S.%%06u [hotstuff info] %%.6f\n", tmp);
- fprintf(stderr, fmt, e.first.tv_usec, e.second);
- }
-#endif
+ for (const auto &e: elapsed)
+ {
+ char fmt[64];
+ struct tm *tmp = localtime(&e.first.tv_sec);
+ strftime(fmt, sizeof fmt, "%Y-%m-%d %H:%M:%S.%%06u [hotstuff info] %%.6f\n", tmp);
+ fprintf(stderr, fmt, e.first.tv_usec, e.second);
}
+#endif
return 0;
}