WIP: pacemaker clean up

7 files changed, 135 insertions, 206 deletions

diff --git a/hotstuff.conf b/hotstuff.conf
index 2a6cc9f..cadf3ae 100644
--- a/hotstuff.conf
+++ b/hotstuff.conf
@@ -1,5 +1,5 @@
 block-size = 1
-pace-maker = sticky
+pace-maker = rr
 replica = 127.0.0.1:10000;20000, 039f89215177475ac408d079b45acef4591fc477dd690f2467df052cf0c7baba23, 542865a568784c4e77c172b82e99cb8a1a53b7bee5f86843b04960ea4157f420
 replica = 127.0.0.1:10001;20001, 0278740a5bec75e333b3c93965b1609163b15d2e3c2fdef141d4859ec70c238e7a, c261250345ebcd676a0edeea173526608604f626b2e8bc4fd2142d3bde1d44d5
 replica = 127.0.0.1:10002;20002, 0269eb606576a315a630c2483deed35cc4bd845abae1c693f97c440c89503fa92e, 065b010aed5629edfb5289e8b22fc6cc6b33c4013bfdd128caba80c3c02d6d78
diff --git a/include/hotstuff/consensus.h b/include/hotstuff/consensus.h
index 61d9167..e0f2ecc 100644
--- a/include/hotstuff/consensus.h
+++ b/include/hotstuff/consensus.h
@@ -102,7 +102,7 @@ class HotStuffCore {
     /** Call to submit new commands to be decided (executed). "Parents" must
      * contain at least one block, and the first block is the actual parent,
      * while the others are uncles/aunts */
-    void on_propose(const std::vector<uint256_t> &cmds,
+    block_t on_propose(const std::vector<uint256_t> &cmds,
                     const std::vector<block_t> &parents,
                     bytearray_t &&extra = bytearray_t());
 
@@ -115,6 +115,7 @@ class HotStuffCore {
     protected:
     /** Called by HotStuffCore upon the decision being made for cmd. */
     virtual void do_decide(Finality &&fin) = 0;
+    virtual void do_consensus(const block_t &blk) = 0;
     /** Called by HotStuffCore upon broadcasting a new proposal.
      * The user should send the proposal message to all replicas except for
      * itself. */
diff --git a/include/hotstuff/hotstuff.h b/include/hotstuff/hotstuff.h
index 680abce..ffc5e3d 100644
--- a/include/hotstuff/hotstuff.h
+++ b/include/hotstuff/hotstuff.h
@@ -195,6 +195,7 @@ class HotStuffBase: public HotStuffCore {
     void do_broadcast_proposal(const Proposal &) override;
     void do_vote(ReplicaID, const Vote &) override;
     void do_decide(Finality &&) override;
+    void do_consensus(const block_t &blk) override;
 
     protected:
 
diff --git a/include/hotstuff/liveness.h b/include/hotstuff/liveness.h
index 03b375e..cdefdeb 100644
--- a/include/hotstuff/liveness.h
+++ b/include/hotstuff/liveness.h
@@ -51,31 +51,31 @@ class PaceMaker {
     virtual promise_t beat_resp(ReplicaID last_proposer) = 0;
     /** Impeach the current proposer. */
     virtual void impeach() {}
+    virtual void on_consensus(const block_t &) {}
     virtual size_t get_pending_size() = 0;
 };
 
 using pacemaker_bt = BoxObj<PaceMaker>;
 
-/** Parent selection implementation for PaceMaker: select all parents.
- * PaceMakers derived from this class will select the highest block as the
- * direct parent, while including other tail blocks (up to parent_limit) as
- * uncles/aunts. */
-class PMAllParents: public virtual PaceMaker {
+/** Parent selection implementation for PaceMaker: select the highest tail that
+ * follows the current hqc block. */
+class PMHighTail: public virtual PaceMaker {
     block_t hqc_tail;
     const int32_t parent_limit;         /**< maximum number of parents */
+
+    bool check_ancestry(const block_t &_a, const block_t &_b) {
+        block_t b;
+        for (b = _b;
+            b->get_height() > _a->get_height();
+            b = b->get_parents()[0]);
+        return b == _a;
+    }
     
     void reg_hqc_update() {
         hsc->async_hqc_update().then([this](const block_t &hqc) {
-            const auto &pref = hqc;
-            for (const auto &blk: hsc->get_tails())
-            {
-                block_t b;
-                for (b = blk;
-                    b->get_height() > pref->get_height();
-                    b = b->get_parents()[0]);
-                if (b == pref && blk->get_height() > hqc_tail->get_height())
-                    hqc_tail = blk;
-            }
+            for (const auto &tail: hsc->get_tails())
+                if (check_ancestry(hqc, tail) && tail->get_height() > hqc_tail->get_height())
+                    hqc_tail = tail;
             reg_hqc_update();
         });
     }
@@ -89,20 +89,16 @@ class PMAllParents: public virtual PaceMaker {
 
     void reg_receive_proposal() {
         hsc->async_wait_receive_proposal().then([this](const Proposal &prop) {
-            const auto &pref = hsc->get_hqc();
+            const auto &hqc = hsc->get_hqc();
             const auto &blk = prop.blk;
-            block_t b;
-            for (b = blk;
-                b->get_height() > pref->get_height();
-                b = b->get_parents()[0]);
-            if (b == pref && blk->get_height() > hqc_tail->get_height())
+            if (check_ancestry(hqc, blk) && blk->get_height() > hqc_tail->get_height())
                 hqc_tail = blk;
             reg_receive_proposal();
         });
     }
 
     public:
-    PMAllParents(int32_t parent_limit): parent_limit(parent_limit) {}
+    PMHighTail(int32_t parent_limit): parent_limit(parent_limit) {}
     void init() {
         hqc_tail = hsc->get_genesis();
         reg_hqc_update();
@@ -113,19 +109,20 @@ class PMAllParents: public virtual PaceMaker {
     std::vector<block_t> get_parents() override {
         const auto &tails = hsc->get_tails();
         std::vector<block_t> parents{hqc_tail};
-        auto nparents = tails.size();
-        if (parent_limit > 0)
-            nparents = std::min(nparents, (size_t)parent_limit);
-        nparents--;
-        /* add the rest of tails as "uncles/aunts" */
-        for (const auto &blk: tails)
-        {
-            if (blk != hqc_tail)
-            {
-                parents.push_back(blk);
-                if (!--nparents) break;
-            }
-        }
+        // TODO: inclusive block chain
+        // auto nparents = tails.size();
+        // if (parent_limit > 0)
+        //     nparents = std::min(nparents, (size_t)parent_limit);
+        // nparents--;
+        // /* add the rest of tails as "uncles/aunts" */
+        // for (const auto &blk: tails)
+        // {
+        //     if (blk != hqc_tail)
+        //     {
+        //         parents.push_back(blk);
+        //         if (!--nparents) break;
+        //     }
+        // }
         return std::move(parents);
     }
 };
@@ -195,12 +192,12 @@ class PMWaitQC: public virtual PaceMaker {
 };
 
 /** Naive PaceMaker where everyone can be a proposer at any moment. */
-struct PaceMakerDummy: public PMAllParents, public PMWaitQC {
+struct PaceMakerDummy: public PMHighTail, public PMWaitQC {
     PaceMakerDummy(int32_t parent_limit):
-        PMAllParents(parent_limit), PMWaitQC() {}
+        PMHighTail(parent_limit), PMWaitQC() {}
     void init(HotStuffCore *hsc) override {
         PaceMaker::init(hsc);
-        PMAllParents::init();
+        PMHighTail::init();
         PMWaitQC::init();
     }
 };
@@ -489,13 +486,13 @@ class PMStickyProposer: virtual public PaceMaker {
     }
 };
 
-struct PaceMakerSticky: public PMAllParents, public PMStickyProposer {
+struct PaceMakerSticky: public PMHighTail, public PMStickyProposer {
     PaceMakerSticky(int32_t parent_limit, double qc_timeout, EventContext eb):
-        PMAllParents(parent_limit), PMStickyProposer(qc_timeout, eb) {}
+        PMHighTail(parent_limit), PMStickyProposer(qc_timeout, eb) {}
 
     void init(HotStuffCore *hsc) override {
         PaceMaker::init(hsc);
-        PMAllParents::init();
+        PMHighTail::init();
         PMStickyProposer::init();
     }
 };
@@ -504,69 +501,41 @@ struct PaceMakerSticky: public PMAllParents, public PMStickyProposer {
  * Simple long-standing round-robin style proposer liveness gadget.
  */
 class PMRoundRobinProposer: virtual public PaceMaker {
-    enum {
-        PROPOSER,
-        FOLLOWER,
-        CANDIDATE /* rotating */
-    } role;
     double qc_timeout;
-    double candidate_timeout;
+    double base_timeout;
+    double exp_timeout;
+    double prop_delay;
     EventContext ec;
     /** QC timer or randomized timeout */
     TimerEvent timer;
-    TimerEvent ev_imp;
-    block_t last_proposed;
     /** the proposer it believes */
     ReplicaID proposer;
+    std::unordered_map<ReplicaID, block_t> prop_blk;
+    bool rotating;
 
     /* extra state needed for a proposer */
     std::queue<promise_t> pending_beats;
+    block_t last_proposed;
     bool locked;
-
-    /* extra state needed for a candidate */
-    std::unordered_map<ReplicaID, promise_t> last_proposed_by;
-
-    promise_t pm_wait_receive_proposal;
-    promise_t pm_wait_propose;
     promise_t pm_qc_finish;
+    promise_t pm_wait_propose;
 
-    void clear_promises() {
-        pm_wait_receive_proposal.reject();
-        pm_wait_propose.reject();
-        pm_qc_finish.reject();
-        for (auto &p: last_proposed_by)
-            p.second.reject();
-        last_proposed_by.clear();
-    }
-
-    /* helper functions for a follower */
-
-    void reg_follower_receive_proposal() {
-        pm_wait_receive_proposal.reject();
-        (pm_wait_receive_proposal = hsc->async_wait_receive_proposal())
-            .then(
-                salticidae::generic_bind(
-                    &PMRoundRobinProposer::follower_receive_proposal, this, _1));
+    void reg_proposal() {
+        hsc->async_wait_proposal().then([this](const Proposal &prop) {
+            auto &pblk = prop_blk[hsc->get_id()];
+            if (!pblk) pblk = prop.blk;
+            if (rotating) reg_proposal();
+        });
     }
 
-    void follower_receive_proposal(const Proposal &prop) {
-        if (prop.proposer == proposer)
-        {
-            auto &qc_ref = prop.blk->get_qc_ref();
-            if (last_proposed && qc_ref != last_proposed)
-            {
-                HOTSTUFF_LOG_INFO("proposer misbehave");
-                to_candidate(); /* proposer misbehave */
-                return;
-            }
-            HOTSTUFF_LOG_PROTO("proposer emits new QC");
-            last_proposed = prop.blk;
-        }
-        reg_follower_receive_proposal();
+    void reg_receive_proposal() {
+        hsc->async_wait_receive_proposal().then([this](const Proposal &prop) {
+            auto &pblk = prop_blk[prop.proposer];
+            if (!pblk) pblk = prop.blk;
+            if (rotating) reg_receive_proposal();
+        });
     }
 
-    /* helper functions for a proposer */
-
     void proposer_schedule_next() {
         if (!pending_beats.empty() && !locked)
         {
@@ -575,147 +544,100 @@ class PMRoundRobinProposer: virtual public PaceMaker {
             pm_qc_finish.reject();
             (pm_qc_finish = hsc->async_qc_finish(last_proposed))
                 .then([this, pm]() {
-                    timer.del();
+                    HOTSTUFF_LOG_PROTO("got QC, propose a new block");
                     pm.resolve(proposer);
-                    timer.add(qc_timeout);
-                    HOTSTUFF_LOG_PROTO("QC timer reset");
                 });
             locked = true;
         }
     }
 
-    void reg_proposer_propose() {
+    void proposer_update_last_proposed() {
         pm_wait_propose.reject();
         (pm_wait_propose = hsc->async_wait_proposal()).then(
-            salticidae::generic_bind(
-                &PMRoundRobinProposer::proposer_propose, this, _1));
+                [this](const Proposal &prop) {
+            last_proposed = prop.blk;
+            locked = false;
+            proposer_schedule_next();
+            proposer_update_last_proposed();
+        });
     }
 
-    void proposer_propose(const Proposal &prop) {
-        last_proposed = prop.blk;
-        locked = false;
-        proposer_schedule_next();
-        reg_proposer_propose();
+    void do_new_consensus(int x) {
+        auto dummy = hsc->on_propose(std::vector<uint256_t>{}, get_parents(), bytearray_t());
+        pm_qc_finish.reject();
+        (pm_qc_finish = hsc->async_qc_finish(dummy))
+            .then([this, x]() {
+#ifdef HOTSTUFF_TWO_STEP
+                if (x >= 2) return;
+#else
+                if (x >= 3) return;
+#endif
+                HOTSTUFF_LOG_PROTO("Pacemaker: got QC for dummy block %d", x);
+                do_new_consensus(x + 1);
+            });
     }
 
-    void propose_elect_block() {
-        DataStream s;
-        /* FIXME: should extra data be the voter's id? */
-        s << hsc->get_id();
-        /* propose a block for leader election */
-        hsc->on_propose(std::vector<uint256_t>{},
-                get_parents(), std::move(s));
+    void on_exp_timeout(TimerEvent &) {
+        if (proposer == hsc->get_id())
+            do_new_consensus(0);
+        timer = TimerEvent(ec, [this](TimerEvent &){ rotate(); });
+        timer.add(prop_delay);
     }
 
-    /* helper functions for a candidate */
-
-    void reg_cp_receive_proposal() {
-        pm_wait_receive_proposal.reject();
-        (pm_wait_receive_proposal = hsc->async_wait_receive_proposal())
-            .then(
-                salticidae::generic_bind(
-                    &PMRoundRobinProposer::cp_receive_proposal, this, _1));
-    }
+    /* role transitions */
 
-    void cp_receive_proposal(const Proposal &prop) {
-        auto _proposer = prop.proposer;
-        auto &p = last_proposed_by[_proposer];
-        HOTSTUFF_LOG_PROTO("got block %s from %d", std::string(*prop.blk).c_str(), _proposer);
-        p.reject();
-        (p = hsc->async_qc_finish(prop.blk)).then([this, blk=prop.blk, _proposer]() {
-            if (_proposer == proposer)
-                to_follower();
-        });
-        reg_cp_receive_proposal();
+    void rotate() {
+        reg_proposal();
+        reg_receive_proposal();
+        prop_blk.clear();
+        rotating = true;
+        proposer = (proposer + 1) % hsc->get_config().nreplicas;
+        HOTSTUFF_LOG_PROTO("Pacemaker: rotate to %d", proposer);
+        pm_qc_finish.reject();
+        pm_wait_propose.reject();
+        // start timer
+        timer = TimerEvent(ec, salticidae::generic_bind(&PMRoundRobinProposer::on_exp_timeout, this, _1));
+        timer.add(exp_timeout);
+        exp_timeout *= 2;
     }
 
-    void candidate_qc_timeout() {
+    void stop_rotate() {
         timer.del();
-        timer.add(candidate_timeout);
-        candidate_timeout *= 1.01;
-        proposer = (proposer + 1) % hsc->get_config().nreplicas;
-        if (proposer == hsc->get_id())
-        {
-            pm_qc_finish.reject();
-            pm_wait_propose.reject();
-            (pm_wait_propose = hsc->async_wait_proposal()).then([this](const Proposal &prop) {
-                const auto &blk = prop.blk;
-                (pm_qc_finish = hsc->async_qc_finish(blk)).then([this, blk]() {
-                    HOTSTUFF_LOG_INFO("collected QC for %s", std::string(*blk).c_str());
-                    /* managed to collect a QC */
-                    to_proposer();
-                    propose_elect_block();
-                });
-            });
-            propose_elect_block();
-        }
-        HOTSTUFF_LOG_INFO("candidate rotates to %d, next try in %.2fs",
-                            proposer, candidate_timeout);
+        HOTSTUFF_LOG_PROTO("Pacemaker: stop rotation at %d", proposer);
+        pm_qc_finish.reject();
+        pm_wait_propose.reject();
+        rotating = false;
+        locked = false;
+        last_proposed = hsc->get_genesis();
+        proposer_update_last_proposed();
     }
 
-    /* role transitions */
-
-    void to_follower() {
-        HOTSTUFF_LOG_INFO("new role: follower");
-        clear_promises();
-        role = FOLLOWER;
-        last_proposed = nullptr;
-        hsc->set_vote_disabled(false);
-        timer.clear();
-        /* redirect all pending cmds to the new proposer */
-        while (!pending_beats.empty())
+    protected:
+    void on_consensus(const block_t &blk) override {
+        if (!rotating)
         {
-            pending_beats.front().resolve(proposer);
-            pending_beats.pop();
+            timer.del();
+            exp_timeout = base_timeout;
         }
-        reg_follower_receive_proposal();
-    }
-
-    void to_proposer() {
-        HOTSTUFF_LOG_INFO("new role: proposer");
-        clear_promises();
-        role = PROPOSER;
-        last_proposed = nullptr;
-        hsc->set_vote_disabled(true);
-        timer = TimerEvent(ec, [this](TimerEvent &) {
-            /* proposer unable to get a QC in time */
-            to_candidate();
-        });
-        proposer_propose(Proposal(-1, hsc->get_genesis(), nullptr));
-    }
-
-    void to_candidate() {
-        HOTSTUFF_LOG_INFO("new role: candidate");
-        clear_promises();
-        role = CANDIDATE;
-        last_proposed = nullptr;
-        hsc->set_vote_disabled(false);
-        timer = TimerEvent(ec, [this](TimerEvent &) {
-            candidate_qc_timeout();
-        });
-        candidate_timeout = qc_timeout * 0.1;
-        reg_cp_receive_proposal();
-        candidate_qc_timeout();
+        if (prop_blk[proposer] == blk)
+            stop_rotate();
     }
 
-    protected:
     void impeach() override {
-        if (role == CANDIDATE) return;
-        ev_imp = TimerEvent(ec, [this](TimerEvent &) {
-            to_candidate();
-        });
-        ev_imp.add(0);
+        if (rotating) return;
+        rotate();
         HOTSTUFF_LOG_INFO("schedule to impeach the proposer");
     }
 
     public:
     PMRoundRobinProposer(double qc_timeout, const EventContext &ec):
-        qc_timeout(qc_timeout), ec(ec), proposer(0) {}
+        qc_timeout(qc_timeout), base_timeout(1), prop_delay(1), ec(ec), proposer(0), rotating(false) {}
 
     size_t get_pending_size() override { return pending_beats.size(); }
 
     void init() {
-        to_candidate();
+        exp_timeout = base_timeout;
+        stop_rotate();
     }
 
     ReplicaID get_proposer() override {
@@ -723,12 +645,11 @@ class PMRoundRobinProposer: virtual public PaceMaker {
     }
 
     promise_t beat() override {
-        if (role != FOLLOWER)
+        if (!rotating && proposer == hsc->get_id())
         {
             promise_t pm;
             pending_beats.push(pm);
-            if (role == PROPOSER)
-                proposer_schedule_next();
+            proposer_schedule_next();
             return std::move(pm);
         }
         else
@@ -744,13 +665,13 @@ class PMRoundRobinProposer: virtual public PaceMaker {
     }
 };
 
-struct PaceMakerRR: public PMAllParents, public PMRoundRobinProposer {
+struct PaceMakerRR: public PMHighTail, public PMRoundRobinProposer {
     PaceMakerRR(int32_t parent_limit, double qc_timeout, EventContext eb):
-        PMAllParents(parent_limit), PMRoundRobinProposer(qc_timeout, eb) {}
+        PMHighTail(parent_limit), PMRoundRobinProposer(qc_timeout, eb) {}
 
     void init(HotStuffCore *hsc) override {
         PaceMaker::init(hsc);
-        PMAllParents::init();
+        PMHighTail::init();
         PMRoundRobinProposer::init();
     }
 };
diff --git a/src/consensus.cpp b/src/consensus.cpp
index 5dadbe0..70c1876 100644
--- a/src/consensus.cpp
+++ b/src/consensus.cpp
@@ -142,6 +142,7 @@ void HotStuffCore::update(const block_t &nblk) {
     {
         const block_t &blk = *it;
         blk->decision = 1;
+        do_consensus(blk);
         LOG_PROTO("commit %s", std::string(*blk).c_str());
         for (size_t i = 0; i < blk->cmds.size(); i++)
             do_decide(Finality(id, 1, i, blk->height,
@@ -150,7 +151,7 @@ void HotStuffCore::update(const block_t &nblk) {
     b_exec = blk;
 }
 
-void HotStuffCore::on_propose(const std::vector<uint256_t> &cmds,
+block_t HotStuffCore::on_propose(const std::vector<uint256_t> &cmds,
                             const std::vector<block_t> &parents,
                             bytearray_t &&extra) {
     if (parents.empty())
@@ -189,6 +190,7 @@ void HotStuffCore::on_propose(const std::vector<uint256_t> &cmds,
     on_propose_(prop);
     /* boradcast to other replicas */
     do_broadcast_proposal(prop);
+    return bnew;
 }
 
 void HotStuffCore::on_receive_proposal(const Proposal &prop) {
@@ -249,8 +251,8 @@ void HotStuffCore::on_receive_vote(const Vote &vote) {
     if (qsize + 1 == config.nmajority)
     {
         qc->compute();
-        on_qc_finish(blk);
         update_hqc(blk, qc);
+        on_qc_finish(blk);
     }
 }
 /*** end HotStuff protocol logic ***/
diff --git a/src/hotstuff.cpp b/src/hotstuff.cpp
index 12a44dc..22f0d82 100644
--- a/src/hotstuff.cpp
+++ b/src/hotstuff.cpp
@@ -374,6 +374,10 @@ void HotStuffBase::do_vote(ReplicaID last_proposer, const Vote &vote) {
     });
 }
 
+void HotStuffBase::do_consensus(const block_t &blk) {
+    pmaker->on_consensus(blk);
+}
+
 void HotStuffBase::do_decide(Finality &&fin) {
     part_decided++;
     state_machine_execute(fin);
diff --git a/src/hotstuff_app.cpp b/src/hotstuff_app.cpp
index 7aa9e1d..778c195 100644
--- a/src/hotstuff_app.cpp
+++ b/src/hotstuff_app.cpp
@@ -377,7 +377,7 @@ void HotStuffApp::start(const std::vector<std::tuple<NetAddr, bytearray_t, bytea
         get_pace_maker().impeach();
         reset_imp_timer();
     });
-    impeach_timer.add(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());