<commit_before>// This file is part of the "x0" project, http://github.com/christianparpart/x0>
//   (c) 2009-2017 <NAME> <<EMAIL>>
//
// Licensed under the MIT License (the "License"); you may not use this
// file except in compliance with the License. You may obtain a copy of
// the License at: http://opensource.org/licenses/MIT

#include <xzero/raft/Server.h>
#include <xzero/raft/ServerUtil.h>
#include <xzero/raft/Discovery.h>
#include <xzero/raft/Error.h>
#include <xzero/raft/StateMachine.h>
#include <xzero/raft/Storage.h>
#include <xzero/raft/Transport.h>
#include <xzero/MonotonicClock.h>
#include <xzero/StringUtil.h>
#include <xzero/logging.h>
#include <system_error>
#include <algorithm>

namespace xzero {

template<>
std::string StringUtil::toString(raft::ServerState s) {
  switch (s) {
    case raft::ServerState::Follower:
      return "Follower";
    case raft::ServerState::Candidate:
      return "Candidate";
    case raft::ServerState::Leader:
      return "Leader";
  }
}

namespace raft {

Server::Server(Id id,
               Storage* storage,
               const Discovery* discovery,
               Transport* transport,
               StateMachine* sm)
      : Server(id, storage, discovery, transport, sm,
               5,                   // maxCommandsPerMessage
               1024,                // maxCommandsSizePerMessage
               500_milliseconds,    // heartbeatTimeout
               250_milliseconds,    // electionTimeout
               500_milliseconds) {  // commitTimeout
}

Server::Server(Id id,
               Storage* storage,
               const Discovery* discovery,
               Transport* transport,
               StateMachine* sm,
               size_t maxCommandsPerMessage,
               size_t maxCommandsSizePerMessage,
               Duration heartbeatTimeout,
               Duration electionTimeout,
               Duration commitTimeout)
    : executor_(std::make_unique<CatchAndLogExceptionHandler>("raft")),
      id_(id),
      currentLeaderId_(0),
      storage_(storage),
      discovery_(discovery),
      transport_(transport),
      stateMachine_(sm),
      state_(ServerState::Follower),
      nextHeartbeat_(MonotonicClock::now()),
      timer_(&executor_, std::bind(&Server::onTimeout, this)),
      verifyLeaderCallbacks_(),
      heartbeatTimeout_(heartbeatTimeout),
      electionTimeout_(electionTimeout),
      commitTimeout_(commitTimeout),
      maxCommandsPerMessage_(maxCommandsPerMessage),
      maxCommandsSizePerMessage_(maxCommandsSizePerMessage),
      running_(false),
      commitIndex_(0),
      lastApplied_(0),
      votesGranted_(0),
      appliedPromises_(),
      onStateChanged(),
      onLeaderChanged() {
  transport_->setHandler(this);
}

Server::~Server() {
}

size_t Server::quorum() const {
  return discovery_->totalMemberCount() / 2 + 1;
}

std::error_code Server::start() {
  std::error_code ec = storage_->initialize(&id_);
  if (ec)
    return ec;

  logDebug("raft.Server",
           "Server $0 starts with term $1 and index $2",
           id(), currentTerm(), commitIndex());

  executor_.execute(StringUtil::format("apply/$0", id_),
                    std::bind(&Server::applyLogsLoop, this));

  running_ = true;

  timer_.setTimeout(ServerUtil::cumulativeDuration(heartbeatTimeout_));
  timer_.start();

  return ec;
}

std::error_code Server::startWithLeader(Id leaderId) {
  std::error_code ec = storage_->initialize(&id_);
  if (ec)
    return ec;

  executor_.execute(StringUtil::format("apply/$0", id_),
                    std::bind(&Server::applyLogsLoop, this));

  running_ = true;

  if (leaderId == id_) {
    logDebug("raft.Server",
             "Server $0 starts with term $1 and index $2 (as leader)",
             id(), currentTerm(), commitIndex(), leaderId);

    setupLeader();
  } else {
    logDebug("raft.Server",
             "Server $0 starts with term $1 and index $2 (as follower)",
             id(), currentTerm(), commitIndex(), leaderId);

    currentLeaderId_ = id_;

    timer_.setTimeout(ServerUtil::cumulativeDuration(heartbeatTimeout_));
    timer_.start();
  }

  return ec;
}

void Server::stop() {
  logDebug("raft.Server", "$0 Sending STOP event.", id_);
  timer_.cancel();
  running_ = false;
  shutdownWakeup_.wakeup();
}

void Server::waitUntilStopped() {
  if (running_) {
    logDebug("raft.Server", "$0 waitUntilStopped: wait for wakeup", id_);
    shutdownWakeup_.waitForFirstWakeup();
    logDebug("raft.Server", "$0 waitUntilStopped: stop-event received", id_);
  } else {
    logDebug("raft.Server", "$0 waitUntilStopped: already stopped", id_);
  }
}

Result<Reply> Server::sendCommand(Command&& command) {
  return sendCommandAsync(std::move(command)).waitAndGetResult();
}

Future<Reply> Server::sendCommandAsync(Command&& command) {
  Promise<Reply> promise;

  if (state_ != ServerState::Leader) {
    promise.failure(std::make_error_code(RaftError::NotLeading));
    return promise.future();
  }

  LogEntry entry(currentTerm(), std::move(command));

  Future<Index> localAck = storage_->appendLogEntryAsync(entry);

  localAck.onFailure(promise);
  localAck.onSuccess([this, promise](const Index& index) {
    logDebug("raft.Server", "$0 log entry locally applied. Wakup replication threads", id());
    appliedPromises_.insert(std::make_pair(index, promise));

    for (Id peerId: discovery_->listMembers()) {
      if (peerId != id_) {
        wakeupReplicationTo(peerId);
      }
    }
  });

  return promise.future();
}

void Server::verifyLeader(std::function<void(bool)> callback) {
  if (state_ != ServerState::Leader) {
    callback(false);
  } else if (nextHeartbeat_ < MonotonicClock::now()) {
    callback(true);
  } else {
    verifyLeaderCallbacks_.emplace_back(callback);
  }
}

void Server::sendVoteRequest() {
  assert(state_ == ServerState::Candidate && "Must be in CANDIDATE state to vote.");

  timer_.setTimeout(ServerUtil::alleviatedDuration(electionTimeout_));
  timer_.rewind();

  votesGranted_ = 0;
  setCurrentTerm(currentTerm() + 1);
  storage_->setVotedFor(id_, currentTerm());

  VoteRequest voteRequest{
    .term         = currentTerm(),
    .candidateId  = id_,
    .lastLogIndex = latestIndex(),
    .lastLogTerm  = getLogTerm(latestIndex()),
  };

  for (Id peerId: discovery_->listMembers()) {
    if (peerId != id_) {
      executor_.execute(StringUtil::format("voter/$0", peerId),
                        [this, peerId, voteRequest]() {
                          transport_->send(peerId, voteRequest);
                        });
    }
  }
}

void Server::onTimeout() {
  if (!running_)
    return;

  switch (state_) {
    case ServerState::Follower:
      logWarning("raft.Server", "$0: Timed out waiting for heartbeat from leader. [$1]", id_, timer_.timeout());
      setState(ServerState::Candidate);
      sendVoteRequest();
      break;<commit_msg>[xzero] raft: also lock on timeout in Server object<commit_after>// This file is part of the "x0" project, http://github.com/christianparpart/x0>
//   (c) 2009-2017 <NAME> <<EMAIL>>
//
// Licensed under the MIT License (the "License"); you may not use this
// file except in compliance with the License. You may obtain a copy of
// the License at: http://opensource.org/licenses/MIT

#include <xzero/raft/Server.h>
#include <xzero/raft/ServerUtil.h>
#include <xzero/raft/Discovery.h>
#include <xzero/raft/Error.h>
#include <xzero/raft/StateMachine.h>
#include <xzero/raft/Storage.h>
#include <xzero/raft/Transport.h>
#include <xzero/MonotonicClock.h>
#include <xzero/StringUtil.h>
#include <xzero/logging.h>
#include <system_error>
#include <algorithm>

namespace xzero {

template<>
std::string StringUtil::toString(raft::ServerState s) {
  switch (s) {
    case raft::ServerState::Follower:
      return "Follower";
    case raft::ServerState::Candidate:
      return "Candidate";
    case raft::ServerState::Leader:
      return "Leader";
  }
}

namespace raft {

Server::Server(Id id,
               Storage* storage,
               const Discovery* discovery,
               Transport* transport,
               StateMachine* sm)
      : Server(id, storage, discovery, transport, sm,
               5,                   // maxCommandsPerMessage
               1024,                // maxCommandsSizePerMessage
               500_milliseconds,    // heartbeatTimeout
               250_milliseconds,    // electionTimeout
               500_milliseconds) {  // commitTimeout
}

Server::Server(Id id,
               Storage* storage,
               const Discovery* discovery,
               Transport* transport,
               StateMachine* sm,
               size_t maxCommandsPerMessage,
               size_t maxCommandsSizePerMessage,
               Duration heartbeatTimeout,
               Duration electionTimeout,
               Duration commitTimeout)
    : executor_(std::make_unique<CatchAndLogExceptionHandler>("raft")),
      id_(id),
      currentLeaderId_(0),
      storage_(storage),
      discovery_(discovery),
      transport_(transport),
      stateMachine_(sm),
      state_(ServerState::Follower),
      nextHeartbeat_(MonotonicClock::now()),
      timer_(&executor_, std::bind(&Server::onTimeout, this)),
      verifyLeaderCallbacks_(),
      heartbeatTimeout_(heartbeatTimeout),
      electionTimeout_(electionTimeout),
      commitTimeout_(commitTimeout),
      maxCommandsPerMessage_(maxCommandsPerMessage),
      maxCommandsSizePerMessage_(maxCommandsSizePerMessage),
      running_(false),
      commitIndex_(0),
      lastApplied_(0),
      votesGranted_(0),
      appliedPromises_(),
      onStateChanged(),
      onLeaderChanged() {
  transport_->setHandler(this);
}

Server::~Server() {
}

size_t Server::quorum() const {
  return discovery_->totalMemberCount() / 2 + 1;
}

std::error_code Server::start() {
  std::error_code ec = storage_->initialize(&id_);
  if (ec)
    return ec;

  logDebug("raft.Server",
           "Server $0 starts with term $1 and index $2",
           id(), currentTerm(), commitIndex());

  executor_.execute(StringUtil::format("apply/$0", id_),
                    std::bind(&Server::applyLogsLoop, this));

  running_ = true;

  timer_.setTimeout(ServerUtil::cumulativeDuration(heartbeatTimeout_));
  timer_.start();

  return ec;
}

std::error_code Server::startWithLeader(Id leaderId) {
  std::error_code ec = storage_->initialize(&id_);
  if (ec)
    return ec;

  executor_.execute(StringUtil::format("apply/$0", id_),
                    std::bind(&Server::applyLogsLoop, this));

  running_ = true;

  if (leaderId == id_) {
    logDebug("raft.Server",
             "Server $0 starts with term $1 and index $2 (as leader)",
             id(), currentTerm(), commitIndex(), leaderId);

    setupLeader();
  } else {
    logDebug("raft.Server",
             "Server $0 starts with term $1 and index $2 (as follower)",
             id(), currentTerm(), commitIndex(), leaderId);

    currentLeaderId_ = id_;

    timer_.setTimeout(ServerUtil::cumulativeDuration(heartbeatTimeout_));
    timer_.start();
  }

  return ec;
}

void Server::stop() {
  logDebug("raft.Server", "$0 Sending STOP event.", id_);
  timer_.cancel();
  running_ = false;
  shutdownWakeup_.wakeup();
}

void Server::waitUntilStopped() {
  if (running_) {
    logDebug("raft.Server", "$0 waitUntilStopped: wait for wakeup", id_);
    shutdownWakeup_.waitForFirstWakeup();
    logDebug("raft.Server", "$0 waitUntilStopped: stop-event received", id_);
  } else {
    logDebug("raft.Server", "$0 waitUntilStopped: already stopped", id_);
  }
}

Result<Reply> Server::sendCommand(Command&& command) {
  return sendCommandAsync(std::move(command)).waitAndGetResult();
}

Future<Reply> Server::sendCommandAsync(Command&& command) {
  Promise<Reply> promise;

  if (state_ != ServerState::Leader) {
    promise.failure(std::make_error_code(RaftError::NotLeading));
    return promise.future();
  }

  LogEntry entry(currentTerm(), std::move(command));

  Future<Index> localAck = storage_->appendLogEntryAsync(entry);

  localAck.onFailure(promise);
  localAck.onSuccess([this, promise](const Index& index) {
    logDebug("raft.Server", "$0 log entry locally applied. Wakup replication threads", id());
    appliedPromises_.insert(std::make_pair(index, promise));

    for (Id peerId: discovery_->listMembers()) {
      if (peerId != id_) {
        wakeupReplicationTo(peerId);
      }
    }
  });

  return promise.future();
}

void Server::verifyLeader(std::function<void(bool)> callback) {
  if (state_ != ServerState::Leader) {
    callback(false);
  } else if (nextHeartbeat_ < MonotonicClock::now()) {
    callback(true);
  } else {
    verifyLeaderCallbacks_.emplace_back(callback);
  }
}

void Server::sendVoteRequest() {
  assert(state_ == ServerState::Candidate && "Must be in CANDIDATE state to vote.");

  timer_.setTimeout(ServerUtil::alleviatedDuration(electionTimeout_));
  timer_.rewind();

  votesGranted_ = 0;
  setCurrentTerm(currentTerm() + 1);
  storage_->setVotedFor(id_, currentTerm());

  VoteRequest voteRequest{
    .term         = currentTerm(),
    .candidateId  = id_,
    .lastLogIndex = latestIndex(),
    .lastLogTerm  = getLogTerm(latestIndex()),
  };

  for (Id peerId: discovery_->listMembers()) {
    if (peerId != id_) {
      executor_.execute(StringUtil::format("voter/$0", peerId),
                        [this, peerId, voteRequest]() {
                          transport_->send(peerId, voteRequest);
                        });
    }
  }
}

void Server::onTimeout() {
  std::lock_guard<decltype(serverLock_)> _lk(serverLock_);

  if (!running_)
    return;

  switch (state_) {
    case ServerState::Follower:
      logWarning("raft.Server", "$0: Timed out waiting for heartbeat from leader. [$1]", id_, timer_.timeout());
      setState(ServerState::Candidate);
      sendVoteRequest();
      break;
{'fraction_non_alphanumeric': 0.13555651797314855, 'fraction_numerical': 0.0073624945864010395, 'mean_word_length': 3.2318265119120344, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 68, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2318238', 'n_tokens_mistral': 4221, 'n_tokens_neox': 4077, 'n_words': 865}