text: /**
   * Copyright 2018 VMware
   * Copyright 2018 Ted Yin
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  #ifndef _HOTSTUFF_LIVENESS_H
  #define _HOTSTUFF_LIVENESS_H
  
  #include "salticidae/util.h"
  #include "hotstuff/hotstuff.h"
  
  namespace hotstuff {
  
  using salticidae::_1;
  using salticidae::_2;
  
  /** Abstraction for liveness gadget (oracle). */
  class PaceMaker {
      protected:
      HotStuffCore *hsc;
      public:
      virtual ~PaceMaker() = default;
      /** Initialize the PaceMaker. A derived class should also call the
       * default implementation to set `hsc`. */
      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, the replica should
       * propose the command. */
      virtual promise_t beat() = 0;
      /** Get the current proposer. */
      virtual ReplicaID get_proposer() = 0;
      /** Select the parent blocks for a new block.
       * @return Parent blocks. The block at index 0 is the direct parent, while
       * the others are uncles/aunts. The returned vector should be non-empty. */
      virtual std::vector<block_t> get_parents() = 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
       * */
      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 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) {
              hqc_tail = hqc;
              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();
          });
      }
  
      void reg_proposal() {
          hsc->async_wait_proposal().then([this](const Proposal &prop) {
              hqc_tail = prop.blk;
              reg_proposal();
          });
      }
  
      void reg_receive_proposal() {
          hsc->async_wait_receive_proposal().then([this](const Proposal &prop) {
              const auto &hqc = hsc->get_hqc();
              const auto &blk = prop.blk;
              if (check_ancestry(hqc, blk) && blk->get_height() > hqc_tail->get_height())
                  hqc_tail = blk;
              reg_receive_proposal();
          });
      }
  
      public:
      PMHighTail(int32_t parent_limit): parent_limit(parent_limit) {}
      void init() {
          hqc_tail = hsc->get_genesis();
          reg_hqc_update();
          reg_proposal();
          reg_receive_proposal();
      }
  
      std::vector<block_t> get_parents() override {
          const auto &tails = hsc->get_tails();
          std::vector<block_t> parents{hqc_tail};
          // 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 parents;
      }
  };
  
  /** Beat implementation for PaceMaker: simply wait for the QC of last proposed
   * block.  PaceMakers derived from this class will beat only when the last
   * block proposed by itself gets its QC. */
  class PMWaitQC: public virtual PaceMaker {
      std::queue<promise_t> pending_beats;
      block_t last_proposed;
      bool locked;
      promise_t pm_qc_finish;
      promise_t pm_wait_propose;
  
      protected:
      void schedule_next() {
          if (!pending_beats.empty() && !locked)
          {
              auto pm = pending_beats.front();
              pending_beats.pop();
              pm_qc_finish.reject();
              (pm_qc_finish = hsc->async_qc_finish(last_proposed))
                  .then([this, pm]() {
                      pm.resolve(get_proposer());
                  });
              locked = true;
          }
      }
  
      void update_last_proposed() {
          pm_wait_propose.reject();
          (pm_wait_propose = hsc->async_wait_proposal()).then(
                  [this](const Proposal &prop) {
              last_proposed = prop.blk;
              locked = false;
              schedule_next();
              update_last_proposed();
          });
      }
  
      public:
  
      size_t get_pending_size() override { return pending_beats.size(); }
  
      void init() {
          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);
          schedule_next();
          return pm;
      }
  
      promise_t beat_resp(ReplicaID last_proposer) override {
          return promise_t([last_proposer](promise_t &pm) {
              pm.resolve(last_proposer);
          });
      }
  };
  
  /** Naive PaceMaker where everyone can be a proposer at any moment. */
  struct PaceMakerDummy: public PMHighTail, public PMWaitQC {
      PaceMakerDummy(int32_t parent_limit):
          PMHighTail(parent_limit), PMWaitQC() {}
      void init(HotStuffCore *hsc) override {
          PaceMaker::init(hsc);
          PMHighTail::init();
          PMWaitQC::init();
      }
  };
  
  /** PaceMakerDummy with a fixed proposer. */
  class PaceMakerDummyFixed: public PaceMakerDummy {
      ReplicaID proposer;
  
      public:
      PaceMakerDummyFixed(ReplicaID proposer,
                          int32_t parent_limit):
          PaceMakerDummy(parent_limit),
          proposer(proposer) {}
  
      ReplicaID get_proposer() override {
          return proposer;
      }
  
      promise_t beat_resp(ReplicaID) override {
          return promise_t([this](promise_t &pm) {
              pm.resolve(proposer);
          });
      }
  };
  
  /**
   * Simple long-standing round-robin style proposer liveness gadget.
   */
  class PMRoundRobinProposer: virtual public PaceMaker {
      double base_timeout;
      double exp_timeout;
      double prop_delay;
      EventContext ec;
      /** QC timer or randomized timeout */
      TimerEvent timer;
      /** 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;
      promise_t pm_qc_finish;
      promise_t pm_wait_propose;
      promise_t pm_qc_manual;
  
      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 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();
          });
      }
  
      void proposer_schedule_next() {
          if (!pending_beats.empty() && !locked)
          {
              auto pm = pending_beats.front();
              pending_beats.pop();
              pm_qc_finish.reject();
              (pm_qc_finish = hsc->async_qc_finish(last_proposed))
                  .then([this, pm]() {
                      HOTSTUFF_LOG_PROTO("got QC, propose a new block");
                      pm.resolve(proposer);
                  });
              locked = true;
          }
      }
  
      void proposer_update_last_proposed() {
          pm_wait_propose.reject();
          (pm_wait_propose = hsc->async_wait_proposal()).then(
                  [this](const Proposal &prop) {
              last_proposed = prop.blk;
              locked = false;
              proposer_schedule_next();
              proposer_update_last_proposed();
          });
      }
  
      void do_new_consensus(int x, const std::vector<uint256_t> &cmds) {
          LeaderProposedOrderedList proposed_orderedlist;
          auto blk = hsc->on_propose(get_parents(), proposed_orderedlist, bytearray_t());
          pm_qc_manual.reject();
          (pm_qc_manual = hsc->async_qc_finish(blk))
              .then([this, x]() {
                  HOTSTUFF_LOG_PROTO("Pacemaker: got QC for block %d", x);
  #ifdef HOTSTUFF_TWO_STEP
                  if (x >= 2) return;
  #else
                  if (x >= 3) return;
  #endif
                  do_new_consensus(x + 1, std::vector<uint256_t>{});
              });
      }
  
      void on_exp_timeout(TimerEvent &) {
          if (proposer == hsc->get_id())
              do_new_consensus(0, std::vector<uint256_t>{});
          timer = TimerEvent(ec, [this](TimerEvent &){ rotate(); });
          timer.add(prop_delay);
      }
  
      /* role transitions */
  
      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();
          pm_qc_manual.reject();
          // start timer
          timer = TimerEvent(ec, salticidae::generic_bind(&PMRoundRobinProposer::on_exp_timeout, this, _1));
          timer.add(exp_timeout);
          exp_timeout *= 2;
      }
  
      void stop_rotate() {
          timer.del();
          HOTSTUFF_LOG_PROTO("Pacemaker: stop rotation at %d", proposer);
          pm_qc_finish.reject();
          pm_wait_propose.reject();
          pm_qc_manual.reject();
          rotating = false;
          locked = false;
          last_proposed = hsc->get_genesis();
          proposer_update_last_proposed();
          if (proposer == hsc->get_id())
          {
              auto hs = static_cast<hotstuff::HotStuffBase *>(hsc);
              hs->do_elected();
              hs->get_tcall().async_call([this, hs](salticidae::ThreadCall::Handle &) {
                  auto &pending = hs->get_decision_waiting();
                  if (!pending.size()) return;
                  HOTSTUFF_LOG_PROTO("reproposing pending commands");
                  std::vector<uint256_t> cmds;
                  for (auto &p: pending)
                      cmds.push_back(p.first);
                  do_new_consensus(0, cmds);
              });
          }
      }
  
      protected:
      void on_consensus(const block_t &blk) override {
          timer.del();
          exp_timeout = base_timeout;
          if (prop_blk[proposer] == blk)
              stop_rotate();
      }
  
      void impeach() override {
          if (rotating) return;
          rotate();
          // HOTSTUFF_LOG_INFO("schedule to impeach the proposer");
      }
  
      public:
      PMRoundRobinProposer(const EventContext &ec,
                          double base_timeout, double prop_delay):
          base_timeout(base_timeout),
          prop_delay(prop_delay),
          ec(ec), proposer(0), rotating(false) {}
  
      size_t get_pending_size() override { return pending_beats.size(); }
  
      void init() {
          exp_timeout = base_timeout;
          stop_rotate();
      }
  
      ReplicaID get_proposer() override {
          return proposer;
      }
  
      promise_t beat() override {
          if (!rotating && proposer == hsc->get_id())
          {
              promise_t pm;
              pending_beats.push(pm);
              proposer_schedule_next();
              return pm;
          }
          else
              return promise_t([proposer=proposer](promise_t &pm) {
                  pm.resolve(proposer);
              });
      }
  
      promise_t beat_resp(ReplicaID last_proposer) override {
          return promise_t([this](promise_t &pm) {
              pm.resolve(proposer);
          });
      }
  };
  
  struct PaceMakerRR: public PMHighTail, public PMRoundRobinProposer {
      PaceMakerRR(EventContext ec, int32_t parent_limit,
                  double base_timeout = 1, double prop_delay = 1):
          PMHighTail(parent_limit),
          PMRoundRobinProposer(ec, base_timeout, prop_delay) {}
  
      void init(HotStuffCore *hsc) override {
          PaceMaker::init(hsc);
          PMHighTail::init();
          PMRoundRobinProposer::init();
      }
  };
  
  }
  
  #endif
  