add PaceMakerDummyFixed

11 files changed, 240 insertions, 130 deletions

diff --git a/include/hotstuff/client.h b/include/hotstuff/client.h
index 00ec77d..2127595 100644
--- a/include/hotstuff/client.h
+++ b/include/hotstuff/client.h
@@ -4,6 +4,7 @@
 #include "salticidae/msg.h"
 #include "hotstuff/type.h"
 #include "hotstuff/entity.h"
+#include "hotstuff/consensus.h"
 
 namespace hotstuff {
 
@@ -13,6 +14,39 @@ enum {
     CHK_CMD = 0x6
 };
 
+struct Finality: public Serializable {
+    ReplicaID rid;
+    int8_t decision;
+    uint256_t blk_hash;
+    
+    public:
+    Finality() = default;
+    Finality(ReplicaID rid, int8_t decision, uint256_t blk_hash):
+        rid(rid), decision(decision), blk_hash(blk_hash) {}
+
+    void serialize(DataStream &s) const override {
+        s << rid << decision;
+        if (decision == 1) s << blk_hash;
+    }
+
+    void unserialize(DataStream &s) override {
+        s >> rid >> decision;
+        if (decision == 1) s >> blk_hash;
+    }
+};
+
+struct MsgClient: public salticidae::MsgBase<> {
+    using MsgBase::MsgBase;
+    void gen_reqcmd(const Command &cmd);
+    void parse_reqcmd(command_t &cmd, HotStuffCore *hsc) const;
+
+    void gen_respcmd(const uint256_t &cmd_hash, const Finality &fin);
+    void parse_respcmd(uint256_t &cmd_hash, Finality &fin) const;
+
+    void gen_chkcmd(const uint256_t &cmd_hash);
+    void parse_chkcmd(uint256_t &cmd_hash) const;
+};
+
 class CommandDummy: public Command {
     static uint64_t cnt;
     uint64_t n;
@@ -49,18 +83,6 @@ class CommandDummy: public Command {
     }
 };
 
-struct MsgClient: public salticidae::MsgBase<> {
-    using MsgBase::MsgBase;
-    void gen_reqcmd(const Command &cmd);
-    void parse_reqcmd(CommandDummy &cmd) const;
-
-    void gen_respcmd(const uint256_t &cmd_hash, const Finality &fin);
-    void parse_respcmd(uint256_t &cmd_hash, Finality &fin) const;
-
-    void gen_chkcmd(const uint256_t &cmd_hash);
-    void parse_chkcmd(uint256_t &cmd_hash) const;
-};
-
 }
 
 #endif
diff --git a/include/hotstuff/consensus.h b/include/hotstuff/consensus.h
index 18e891e..cb18db5 100644
--- a/include/hotstuff/consensus.h
+++ b/include/hotstuff/consensus.h
@@ -1,6 +1,7 @@
 #ifndef _HOTSTUFF_CONSENSUS_H
 #define _HOTSTUFF_CONSENSUS_H
 
+#include <cassert>
 #include <set>
 #include <unordered_map>
 
@@ -125,7 +126,7 @@ class HotStuffCore {
     block_t get_genesis() { return b0; }
     const ReplicaConfig &get_config() { return config; }
     int8_t get_cmd_decision(const uint256_t &cmd_hash);
-    ReplicaID get_id() { return id; }
+    ReplicaID get_id() const { return id; }
     operator std::string () const;
 };
 
diff --git a/include/hotstuff/entity.h b/include/hotstuff/entity.h
index 00c64a6..03aff06 100644
--- a/include/hotstuff/entity.h
+++ b/include/hotstuff/entity.h
@@ -73,26 +73,6 @@ class HotStuffCore;
 using block_t = salticidae::RcObj<Block>;
 using block_weak_t = salticidae::WeakObj<Block>;
 
-struct Finality: public Serializable {
-    int8_t decision;
-    uint256_t blk_hash;
-    
-    public:
-    Finality(): decision(0) {}
-    Finality(int8_t decision, uint256_t blk_hash):
-        decision(decision), blk_hash(blk_hash) {}
-
-    void serialize(DataStream &s) const override {
-        s << decision;
-        if (decision == 1) s << blk_hash;
-    }
-
-    void unserialize(DataStream &s) override {
-        s >> decision;
-        if (decision == 1) s >> blk_hash;
-    }
-};
-
 class Command: public Serializable {
     friend HotStuffCore;
     block_weak_t container;
@@ -101,7 +81,6 @@ class Command: public Serializable {
     virtual const uint256_t &get_hash() const = 0;
     virtual bool verify() const = 0;
     inline int8_t get_decision() const;
-    inline Finality get_finality() const;
     block_t get_container() const {
         return container;
     }
@@ -220,12 +199,6 @@ int8_t Command::get_decision() const {
     return cptr ? cptr->get_decision() : 0;
 }
 
-Finality Command::get_finality() const {
-    block_t blk = get_container();
-    return Finality(get_decision(),
-                    blk ? blk->get_hash() : uint256_t());
-}
-
 class EntityStorage {
     std::unordered_map<const uint256_t, block_t> blk_cache;
     std::unordered_map<const uint256_t, command_t> cmd_cache;
diff --git a/include/hotstuff/hotstuff.h b/include/hotstuff/hotstuff.h
index 9546216..9d1a8a8 100644
--- a/include/hotstuff/hotstuff.h
+++ b/include/hotstuff/hotstuff.h
@@ -153,6 +153,7 @@ class HotStuffBase: public HotStuffCore {
     void do_broadcast_proposal(const Proposal &) override;
     void do_vote(ReplicaID, const Vote &) override;
     void do_decide(const command_t &) override;
+    void do_forward(const uint256_t &cmd_hash, ReplicaID rid);
 
     public:
     HotStuffBase(uint32_t blk_size,
@@ -160,30 +161,15 @@ class HotStuffBase: public HotStuffCore {
             ReplicaID rid,
             privkey_bt &&priv_key,
             NetAddr listen_addr,
-            EventContext eb = EventContext(),
-            pacemaker_bt pmaker = nullptr);
+            EventContext eb,
+            pacemaker_bt pmaker);
 
     ~HotStuffBase();
 
     /* the API for HotStuffBase */
 
     /* Submit the command to be decided. */
-    void add_command(command_t cmd) {
-        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);
-            });
-        }
-    }
-
+    ReplicaID add_command(command_t cmd);
     void add_replica(ReplicaID idx, const NetAddr &addr, pubkey_bt &&pub_key);
     void start(bool eb_loop = false);
 
@@ -238,13 +224,15 @@ class HotStuff: public HotStuffBase {
             ReplicaID rid,
             const bytearray_t &raw_privkey,
             NetAddr listen_addr,
-            EventContext eb = nullptr):
+            EventContext eb = EventContext(),
+            pacemaker_bt pmaker = new PaceMakerDummy()):
         HotStuffBase(blk_size,
                     parent_limit,
                     rid,
                     new PrivKeyType(raw_privkey),
                     listen_addr,
-                    eb) {}
+                    eb,
+                    std::move(pmaker)) {}
 
     void add_replica(ReplicaID idx, const NetAddr &addr, const bytearray_t &pubkey_raw) {
         DataStream s(pubkey_raw);
diff --git a/include/hotstuff/liveness.h b/include/hotstuff/liveness.h
index f8d3c50..b23d4c2 100644
--- a/include/hotstuff/liveness.h
+++ b/include/hotstuff/liveness.h
@@ -7,13 +7,16 @@ namespace hotstuff {
 
 /** Abstraction for liveness gadget (oracle). */
 class PaceMaker {
+    protected:
+    HotStuffCore *hsc;
     public:
     virtual ~PaceMaker() = default;
+    virtual void init(HotStuffCore *_hsc) { hsc = _hsc; }
     /** Get a promise resolved when the pace maker thinks it is a *good* time
-     * to issue new commands. When promise is resolved with the ID of itself,
-     * the replica should propose the command, otherwise it will forward the
-     * command to the proposer indicated by the ID. */
+     * to issue new commands. When promise is resolved, the replica should
+     * propose the command. */
     virtual promise_t beat() = 0;
+    virtual ReplicaID get_proposer() = 0;
     /** Get a promise resolved when the pace maker thinks it is a *good* time
      * to vote for a block. The promise is resolved with the next proposer's ID
      * */
@@ -24,7 +27,6 @@ using pacemaker_bt = BoxObj<PaceMaker>;
 
 /** A pace maker that waits for the qc of the last proposed block. */
 class PaceMakerDummy: public PaceMaker {
-    HotStuffCore *hsc;
     std::queue<promise_t> pending_beats;
     block_t last_proposed;
     bool locked;
@@ -34,9 +36,8 @@ class PaceMakerDummy: public PaceMaker {
         {
             auto pm = pending_beats.front();
             pending_beats.pop();
-            hsc->async_qc_finish(last_proposed).then(
-                    [id = hsc->get_id(), pm]() {
-                pm.resolve(id);
+            hsc->async_qc_finish(last_proposed).then([this, pm]() {
+                pm.resolve(get_proposer());
             });
             locked = true;
         }
@@ -52,13 +53,19 @@ class PaceMakerDummy: public PaceMaker {
     }
 
     public:
-    PaceMakerDummy(HotStuffCore *hsc):
-            hsc(hsc),
-            last_proposed(hsc->get_genesis()),
-            locked(false) {
+    PaceMakerDummy() = default;
+
+    void init(HotStuffCore *hsc) override {
+        PaceMaker::init(hsc);
+        last_proposed = hsc->get_genesis();
+        locked = false;
         update_last_proposed();
     }
 
+    ReplicaID get_proposer() override {
+        return hsc->get_id();
+    }
+
     promise_t beat() override {
         promise_t pm;
         pending_beats.push(pm);
@@ -73,6 +80,23 @@ class PaceMakerDummy: public PaceMaker {
     }
 };
 
+class PaceMakerDummyFixed: public PaceMakerDummy {
+    ReplicaID proposer;
+
+    public:
+    ReplicaID get_proposer() override {
+        return proposer;
+    }
+
+    PaceMakerDummyFixed(ReplicaID proposer): proposer(proposer) {}
+
+    promise_t next_proposer(ReplicaID) override {
+        return promise_t([this](promise_t &pm) {
+            pm.resolve(proposer);
+        });
+    }
+};
+
 }
 
 #endif
diff --git a/run_replicas.sh b/run_replicas.sh
index c106d42..213b6b7 100755
--- a/run_replicas.sh
+++ b/run_replicas.sh
@@ -1,5 +1,10 @@
 #!/bin/bash
-for i in {0..3}; do
+rep=({0..3})
+if [[ $# -gt 0 ]]; then
+    rep=($@)
+fi
+for i in "${rep[@]}"; do
+    echo "starting replica $i"
     #valgrind ./hotstuff-app --conf hotstuff-sec${i}.conf > log${i} 2>&1 &
     ./hotstuff-app --conf hotstuff-sec${i}.conf > log${i} 2>&1 &
 done
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());