improve network impl

4 files changed, 63 insertions, 139 deletions

diff --git a/src/consensus.cpp b/src/consensus.cpp
index 688c450..bf4868d 100644
--- a/src/consensus.cpp
+++ b/src/consensus.cpp
@@ -31,7 +31,7 @@ void HotStuffCore::sanity_check_delivered(const block_t &blk) {
         throw std::runtime_error("block not delivered");
 }
 
-block_t HotStuffCore::sanity_check_delivered(const uint256_t &blk_hash) {
+block_t HotStuffCore::get_delivered_blk(const uint256_t &blk_hash) {
     block_t blk = storage->find_blk(blk_hash);
     if (blk == nullptr || !blk->delivered)
         throw std::runtime_error("block not delivered");
@@ -46,10 +46,7 @@ bool HotStuffCore::on_deliver_blk(const block_t &blk) {
     }
     blk->parents.clear();
     for (const auto &hash: blk->parent_hashes)
-    {
-        block_t p = sanity_check_delivered(hash);
-        blk->parents.push_back(p);
-    }
+        blk->parents.push_back(get_delivered_blk(hash));
     blk->height = blk->parents[0]->height + 1;
     for (const auto &cmd: blk->cmds)
         cmd->container = blk;
@@ -66,41 +63,41 @@ bool HotStuffCore::on_deliver_blk(const block_t &blk) {
     tails.insert(blk);
 
     blk->delivered = true;
-    LOG_DEBUG("delivered %.10s", get_hex(blk->get_hash()).c_str());
+    LOG_DEBUG("deliver %s", std::string(*blk).c_str());
     return true;
 }
 
 void HotStuffCore::check_commit(const block_t &_blk) {
     const block_t &blk = _blk->qc_ref;
     if (blk->qc_ref == nullptr) return;
+    /* decided blk could possible be incomplete due to pruning */
     if (blk->decision) return;
     block_t p = blk->parents[0];
-    if (p == blk->qc_ref)
-    { /* commit */
-        std::vector<block_t> commit_queue;
-        block_t b;
-        for (b = p; b->height > bexec->height; b = b->parents[0])
-        { /* todo: also commit the uncles/aunts */
-            commit_queue.push_back(b);
-        }
-        if (b != bexec)
-            throw std::runtime_error("safety breached :(");
-        for (auto it = commit_queue.rbegin(); it != commit_queue.rend(); it++)
-        {
-            const block_t &blk = *it;
-            blk->decision = 1;
+    /* commit requires direct parent */
+    if (p != blk->qc_ref) return;
+    /* otherwise commit */
+    std::vector<block_t> commit_queue;
+    block_t b;
+    for (b = p; b->height > bexec->height; b = b->parents[0])
+    { /* TODO: also commit the uncles/aunts */
+        commit_queue.push_back(b);
+    }
+    if (b != bexec)
+        throw std::runtime_error("safety breached :(");
+    for (auto it = commit_queue.rbegin(); it != commit_queue.rend(); it++)
+    {
+        const block_t &blk = *it;
+        blk->decision = 1;
 #ifdef HOTSTUFF_ENABLE_LOG_PROTO
-            LOG_INFO("commit blk %.10s", get_hex10(blk->get_hash()).c_str());
+        LOG_INFO("commit %s", std::string(*blk).c_str());
 #endif
-            for (auto cmd: blk->cmds)
-                do_decide(cmd);
-        }
-        bexec = p;
+        for (auto cmd: blk->cmds) do_decide(cmd);
     }
+    bexec = p;
 }
 
 bool HotStuffCore::update(const uint256_t &bqc_hash) {
-    block_t _bqc = sanity_check_delivered(bqc_hash);
+    block_t _bqc = get_delivered_blk(bqc_hash);
     if (_bqc->qc_ref == nullptr) return false;
     check_commit(_bqc);
     if (_bqc->qc_ref->height > bqc->qc_ref->height)
@@ -116,13 +113,14 @@ void HotStuffCore::on_propose(const std::vector<command_t> &cmds,
     block_t p = parents[0];
     quorum_cert_bt qc = nullptr;
     block_t qc_ref = nullptr;
+    /* a block can optionally carray a QC */
     if (p != b0 && p->voted.size() >= config.nmajority)
     {
         qc = p->self_qc->clone();
         qc->compute();
         qc_ref = p;
     }
-    /* create a new block */
+    /* create the new block */
     block_t bnew = storage->add_blk(
         Block(
             parents,
@@ -163,8 +161,11 @@ void HotStuffCore::on_receive_proposal(const Proposal &prop) {
         for (b = bnew;
             b->height > pref->height;
             b = b->parents[0]);
-        opinion = b == pref;
-        vheight = bnew->height;
+        if (b == pref) /* on the same branch */
+        {
+            opinion = true;
+            vheight = bnew->height;
+        }
     }
 #ifdef HOTSTUFF_ENABLE_LOG_PROTO
     LOG_INFO("now state: %s", std::string(*this).c_str());
@@ -185,17 +186,17 @@ void HotStuffCore::on_receive_vote(const Vote &vote) {
     LOG_INFO("got %s", std::string(vote).c_str());
     LOG_INFO("now state: %s", std::string(*this).c_str());
 #endif
-
-    block_t blk = sanity_check_delivered(vote.blk_hash);
+    block_t blk = get_delivered_blk(vote.blk_hash);
     if (vote.cert == nullptr) return;
+    /* otherwise the vote is positive */
     if (!vote.verify())
     {
-        LOG_WARN("invalid vote");
+        LOG_WARN("invalid vote from %d", vote.voter);
         return;
     }
     if (!blk->voted.insert(vote.voter).second)
     {
-        LOG_WARN("duplicate votes");
+        LOG_WARN("duplicate vote from %d", vote.voter);
         return;
     }
     size_t qsize = blk->voted.size();
diff --git a/src/hotstuff.cpp b/src/hotstuff.cpp
index d0b42c3..ed15cc1 100644
--- a/src/hotstuff.cpp
+++ b/src/hotstuff.cpp
@@ -411,7 +411,7 @@ HotStuffBase::HotStuffBase(uint32_t blk_size,
     pn.reg_handler(VOTE, std::bind(&HotStuffBase::vote_handler, this, _1, _2));
     pn.reg_handler(QUERY_FETCH_BLK, std::bind(&HotStuffBase::query_fetch_blk_handler, this, _1, _2));
     pn.reg_handler(RESP_FETCH_BLK, std::bind(&HotStuffBase::resp_fetch_blk_handler, this, _1, _2));
-    pn.init(listen_addr);
+    pn.listen(listen_addr);
 }
 
 void HotStuffBase::do_broadcast_proposal(const Proposal &prop) {
diff --git a/src/hotstuff_app.cpp b/src/hotstuff_app.cpp
index e1eec1b..28f02fc 100644
--- a/src/hotstuff_app.cpp
+++ b/src/hotstuff_app.cpp
@@ -221,7 +221,7 @@ HotStuffApp::HotStuffApp(uint32_t blk_size,
     /* register the handlers for msg from clients */
     cn.reg_handler(hotstuff::REQ_CMD, std::bind(&HotStuffApp::client_request_cmd_handler, this, _1, _2));
     cn.reg_handler(hotstuff::CHK_CMD, std::bind(&HotStuffApp::client_check_cmd_handler, this, _1, _2));
-    cn.init(clisten_addr);
+    cn.listen(clisten_addr);
 }
 
 void HotStuffApp::client_request_cmd_handler(const MsgClient &msg, conn_client_t conn_) {
diff --git a/src/hotstuff_client.cpp b/src/hotstuff_client.cpp
index 9478d36..1363f39 100644
--- a/src/hotstuff_client.cpp
+++ b/src/hotstuff_client.cpp
@@ -12,39 +12,22 @@ using salticidae::Config;
 using salticidae::ElapsedTime;
 using salticidae::trim_all;
 using salticidae::split;
+using salticidae::MsgNetwork;
 
 using hotstuff::ReplicaID;
 using hotstuff::NetAddr;
 using hotstuff::EventContext;
-using hotstuff::Event;
 using hotstuff::uint256_t;
-using hotstuff::bytearray_t;
 using hotstuff::MsgClient;
 using hotstuff::CommandDummy;
 using hotstuff::command_t;
 using hotstuff::Finality;
+using hotstuff::HotStuffError;
 
 EventContext eb;
-size_t max_async_num = 10;
-int max_iter_num = 100;
 ReplicaID proposer;
-
-int connect(const NetAddr &node) {
-    int fd;
-    if ((fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == -1)
-        assert(0);
-    struct sockaddr_in sockin;
-    memset(&sockin, 0, sizeof(struct sockaddr_in));
-    sockin.sin_family = AF_INET;
-    sockin.sin_addr.s_addr = node.ip;
-    sockin.sin_port = node.port;
-    if (connect(fd, (struct sockaddr *)&sockin, sizeof(struct sockaddr_in)) == -1)
-        assert(0);
-    HOTSTUFF_LOG_INFO("connected to %s", std::string(node).c_str());
-    return fd;
-}
-
-void on_receive(int);
+size_t max_async_num;
+int max_iter_num;
 
 struct Request {
     ReplicaID rid;
@@ -54,80 +37,18 @@ struct Request {
         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;
+MsgNetwork<MsgClient> mn(eb, 10, 0, 2, 4096);
 
+std::unordered_map<ReplicaID, MsgNetwork<MsgClient>::conn_t> conns;
+std::vector<NetAddr> replicas;
 
-void setup(ReplicaID rid) {
+void set_proposer(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())
-        assert(0);
-}
-
-void read_msg(int fd, MsgClient &msg) {
-    static const size_t BUFF_SEG_SIZE = 1024;
-    ssize_t ret;
-    auto &buffer = buffers[fd];
-    bool read_body = false;
-    for (;;)
-    {
-        bytearray_t buff_seg;
-        if (!read_body && buffer.size() >= MsgClient::header_size)
-        {
-            buff_seg = buffer.pop(MsgClient::header_size);
-            msg = MsgClient(buff_seg.data());
-            read_body = true;
-        }
-        if (read_body && buffer.size() >= msg.get_length())
-        {
-            buff_seg = buffer.pop(msg.get_length());
-            msg.set_payload(std::move(buff_seg));
-            return;
-        }
-
-        buff_seg.resize(BUFF_SEG_SIZE);
-        ret = read(fd, buff_seg.data(), BUFF_SEG_SIZE);
-        assert(ret != -1);
-        if (ret > 0)
-        {
-            buff_seg.resize(ret);
-            buffer.push(std::move(buff_seg));
-        }
-    }
+        conns.insert(std::make_pair(rid, mn.create_conn(replicas[rid])));
 }
 
 void try_send() {
@@ -136,7 +57,7 @@ void try_send() {
         auto cmd = CommandDummy::make_cmd();
         MsgClient msg;
         msg.gen_reqcmd(*cmd);
-        write_msg(conns.find(proposer)->second.fd, msg);
+        mn.send_msg(msg, conns.find(proposer)->second);
         HOTSTUFF_LOG_INFO("send new cmd %.10s",
                             get_hex(cmd->get_hash()).c_str());
         waiting.insert(std::make_pair(
@@ -146,11 +67,9 @@ void try_send() {
     }
 }
 
-void on_receive(int fd) {
-    MsgClient msg;
+void on_receive(const MsgClient &msg, MsgNetwork<MsgClient>::conn_t) {
     uint256_t cmd_hash;
     Finality fin;
-    read_msg(fd, msg);
     HOTSTUFF_LOG_DEBUG("got %s", std::string(msg).c_str());
     if (!msg.verify_checksum())
         HOTSTUFF_LOG_ERROR("incorrect checksum %08x", msg.get_checksum());
@@ -159,13 +78,13 @@ void on_receive(int fd) {
     if (fin.rid != proposer)
     {
         HOTSTUFF_LOG_INFO("reconnect to the new proposer");
-        setup(fin.rid);
+        set_proposer(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);
+        mn.send_msg(msg, conns.find(proposer)->second);
         HOTSTUFF_LOG_INFO("resend cmd %.10s",
                             get_hex(cmd_hash).c_str());
         it->second.et.start();
@@ -173,8 +92,8 @@ void on_receive(int fd) {
         return;
     }
     HOTSTUFF_LOG_INFO(
-        "fd %d got response for %.10s: <decision=%d, blk=%.10s>",
-        fd, get_hex(cmd_hash).c_str(),
+        "got response for %.10s: <decision=%d, blk=%.10s>",
+        get_hex(cmd_hash).c_str(),
         fin.decision,
         get_hex(fin.blk_hash).c_str());
     if (it == waiting.end()) return;
@@ -187,30 +106,34 @@ std::pair<std::string, std::string> split_ip_port_cport(const std::string &s) {
     return std::make_pair(ret[0], ret[1]);
 }
 
-
 int main(int argc, char **argv) {
     Config config("hotstuff.conf");
     auto opt_idx = Config::OptValInt::create(0);
     auto opt_replicas = Config::OptValStrVec::create();
-    auto opt_max_iter_num = Config::OptValInt::create();
+    auto opt_max_iter_num = Config::OptValInt::create(100);
+    auto opt_max_async_num = Config::OptValInt::create(10);
+
+    mn.reg_handler(hotstuff::RESP_CMD, on_receive);
 
     try {
         config.add_opt("idx", opt_idx, Config::SET_VAL);
         config.add_opt("replica", opt_replicas, Config::APPEND);
-        config.add_opt("ntx", opt_max_iter_num, Config::SET_VAL);
+        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 = trim_all(split(s, ","));
-            assert(res.size() == 2);
+            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))
+        if (!(0 <= idx && (size_t)idx < raw.size() && raw.size() > 0))
             throw std::invalid_argument("out of range");
         for (const auto &p: raw)
         {
@@ -219,10 +142,10 @@ int main(int argc, char **argv) {
             replicas.push_back(NetAddr(NetAddr(_p.first).ip, htons(stoi(_p.second, &_))));
         }
 
-        setup(idx);
+        set_proposer(idx);
         try_send();
         eb.dispatch();
-    } catch (hotstuff::HotStuffError &e) {
+    } catch (HotStuffError &e) {
         HOTSTUFF_LOG_ERROR("exception: %s", std::string(e).c_str());
     }
     return 0;