raft: implement utility class for tallying votes

src/CMakeLists.txt

@@ -59,11 +59,12 @@ add_library(XrdQuarkDB SHARED
   raft/RaftTimeouts.cc                    raft/RaftTimeouts.hh
   raft/RaftDirector.cc                    raft/RaftDirector.hh
   raft/RaftCommitTracker.cc               raft/RaftCommitTracker.hh
-  raft/RaftWriteTracker.cc                raft/RaftWriteTracker.hh
   raft/RaftGroup.cc                       raft/RaftGroup.hh
                                           raft/RaftMembers.hh
   raft/RaftTrimmer.cc                     raft/RaftTrimmer.hh
   raft/RaftLease.cc                       raft/RaftLease.hh
+  raft/RaftVoteRegistry.cc                raft/RaftVoteRegistry.hh
+  raft/RaftWriteTracker.cc                raft/RaftWriteTracker.hh
 
   recovery/RecoveryDispatcher.cc          recovery/RecoveryDispatcher.hh
   recovery/RecoveryEditor.cc              recovery/RecoveryEditor.hh

src/raft/RaftCommon.hh

@@ -208,6 +208,9 @@ enum class RaftVote {
 };
 
 struct RaftVoteResponse {
+  RaftVoteResponse(RaftTerm tr, RaftVote vt) : term(tr), vote(vt) {}
+  RaftVoteResponse() : term(0), vote(RaftVote::VETO) {}
+
   RaftTerm term;
   RaftVote vote;
 
@@ -232,6 +235,12 @@ struct RaftVoteResponse {
   }
 };
 
+enum class ElectionOutcome {
+  kElected,
+  kNotElected,
+  kVetoed
+};
+
 inline size_t calculateQuorumSize(size_t members) {
   return (members / 2) + 1;
 }

src/raft/RaftUtils.hh

@@ -48,13 +48,6 @@ public:
   static bool fetchLastResponse(const qclient::redisReplyPtr &source, std::vector<RaftEntry> &entries);
 };
 
-
-enum class ElectionOutcome {
-  kElected,
-  kNotElected,
-  kVetoed
-};
-
 struct ElectionSingleTally {
   bool timeout;
   bool error;

src/raft/RaftVoteRegistry.cc

+// ----------------------------------------------------------------------
+// File: RaftVoteRegistry.cc
+// Author: Georgios Bitzes - CERN
+// ----------------------------------------------------------------------
+
+/************************************************************************
+ * quarkdb - a redis-like highly available key-value store              *
+ * Copyright (C) 2020 CERN/Switzerland                                  *
+ *                                                                      *
+ * This program is free software: you can redistribute it and/or modify *
+ * it under the terms of the GNU General Public License as published by *
+ * the Free Software Foundation, either version 3 of the License, or    *
+ * (at your option) any later version.                                  *
+ *                                                                      *
+ * This program is distributed in the hope that it will be useful,      *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
+ * GNU General Public License for more details.                         *
+ *                                                                      *
+ * You should have received a copy of the GNU General Public License    *
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.*
+ ************************************************************************/
+
+#include "raft/RaftVoteRegistry.hh"
+#include "raft/RaftUtils.hh"
+
+namespace quarkdb {
+
+//------------------------------------------------------------------------------
+// Constructor
+//------------------------------------------------------------------------------
+RaftVoteRegistry::RaftVoteRegistry(RaftTerm term, bool prevote)
+: mTerm(term), mPreVote(prevote) {}
+
+//------------------------------------------------------------------------------
+// Register vote
+//------------------------------------------------------------------------------
+void RaftVoteRegistry::registerVote(const RaftServer &srv, RaftVoteResponse resp) {
+  qdb_assert(mContents.find(srv) == mContents.end());
+
+  SingleVote vote;
+  vote.netError = false;
+  vote.parseError = false;
+  vote.resp = resp;
+
+  mContents[srv] = vote;
+}
+
+//------------------------------------------------------------------------------
+// Register vote
+//------------------------------------------------------------------------------
+void RaftVoteRegistry::registerParseError(const RaftServer &srv) {
+  qdb_assert(mContents.find(srv) == mContents.end());
+
+  SingleVote vote;
+  vote.netError = false;
+  vote.parseError = true;
+
+  mContents[srv] = vote;
+}
+
+//------------------------------------------------------------------------------
+// Register vote
+//------------------------------------------------------------------------------
+void RaftVoteRegistry::registerNetworkError(const RaftServer &srv) {
+  qdb_assert(mContents.find(srv) == mContents.end());
+
+  SingleVote vote;
+  vote.netError = true;
+  vote.parseError = false;
+
+  mContents[srv] = vote;
+}
+
+//------------------------------------------------------------------------------
+// Determine outcome
+//------------------------------------------------------------------------------
+ElectionOutcome RaftVoteRegistry::determineOutcome() const {
+  size_t positives = 0;
+
+  for(auto it = mContents.begin(); it != mContents.end(); it++) {
+    const SingleVote& sv = it->second;
+
+    if(sv.netError) {
+      continue;
+    }
+    else if(sv.parseError) {
+      if(mPreVote) {
+        // Does not support pre-vote... assume granted
+        positives++;
+      }
+    }
+    else if(sv.resp.vote == RaftVote::GRANTED) {
+      positives++;
+    }
+    else if(sv.resp.vote == RaftVote::VETO) {
+      return ElectionOutcome::kVetoed;
+    }
+  }
+
+  // Implicit vote for myself
+  positives++;
+
+  if(positives >= calculateQuorumSize(mContents.size()+1)) {
+    return ElectionOutcome::kElected;
+
+  }
+
+  return ElectionOutcome::kNotElected;
+}
+
+//------------------------------------------------------------------------------
+// Register vote
+//------------------------------------------------------------------------------
+void RaftVoteRegistry::registerVote(const RaftServer &srv, std::future<qclient::redisReplyPtr> &fut, std::chrono::steady_clock::time_point deadline) {
+  if(fut.wait_until(deadline) != std::future_status::ready) {
+    return registerNetworkError(srv);
+  }
+
+  qclient::redisReplyPtr reply = fut.get();
+  if(reply == nullptr) {
+    return registerNetworkError(srv);
+  }
+
+  RaftVoteResponse resp;
+  if(!RaftParser::voteResponse(reply, resp)) {
+    if(!mPreVote) {
+      qdb_critical("Could not parse vote response from " << srv.toString() << ": " << qclient::describeRedisReply(reply));
+    }
+
+    return registerParseError(srv);
+  }
+
+  return registerVote(srv, resp);
+}
+
+}
\ No newline at end of file

src/raft/RaftVoteRegistry.hh

+// ----------------------------------------------------------------------
+// File: RaftVoteRegistry.hh
+// Author: Georgios Bitzes - CERN
+// ----------------------------------------------------------------------
+
+/************************************************************************
+ * quarkdb - a redis-like highly available key-value store              *
+ * Copyright (C) 2020 CERN/Switzerland                                  *
+ *                                                                      *
+ * This program is free software: you can redistribute it and/or modify *
+ * it under the terms of the GNU General Public License as published by *
+ * the Free Software Foundation, either version 3 of the License, or    *
+ * (at your option) any later version.                                  *
+ *                                                                      *
+ * This program is distributed in the hope that it will be useful,      *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
+ * GNU General Public License for more details.                         *
+ *                                                                      *
+ * You should have received a copy of the GNU General Public License    *
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.*
+ ************************************************************************/
+
+#ifndef QUARKDB_RAFT_VOTE_REGISTRY_HH
+#define QUARKDB_RAFT_VOTE_REGISTRY_HH
+
+#include "raft/RaftCommon.hh"
+#include <future>
+#include "qclient/QClient.hh"
+
+namespace quarkdb {
+
+//------------------------------------------------------------------------------
+// Helper class for counting votes received during an election
+//------------------------------------------------------------------------------
+class RaftVoteRegistry {
+public:
+  //----------------------------------------------------------------------------
+  // Hold the response for a single server
+  //----------------------------------------------------------------------------
+  struct SingleVote {
+    bool netError;
+    bool parseError;
+    RaftVoteResponse resp;
+  };
+
+  //----------------------------------------------------------------------------
+  // Constructor
+  //----------------------------------------------------------------------------
+  RaftVoteRegistry(RaftTerm term, bool prevote);
+
+  //----------------------------------------------------------------------------
+  // Register vote
+  //----------------------------------------------------------------------------
+  void registerVote(const RaftServer &srv, RaftVoteResponse resp);
+
+  //----------------------------------------------------------------------------
+  // Register vote
+  //----------------------------------------------------------------------------
+  void registerParseError(const RaftServer &srv);
+
+  //----------------------------------------------------------------------------
+  // Register vote
+  //----------------------------------------------------------------------------
+  void registerNetworkError(const RaftServer &srv);
+
+  //----------------------------------------------------------------------------
+  // Register vote
+  //----------------------------------------------------------------------------
+  void registerVote(const RaftServer &srv, std::future<qclient::redisReplyPtr> &fut,
+    std::chrono::steady_clock::time_point deadline);
+
+  //----------------------------------------------------------------------------
+  // Determine outcome
+  //----------------------------------------------------------------------------
+  ElectionOutcome determineOutcome() const;
+
+
+private:
+  RaftTerm mTerm;
+  bool mPreVote;
+
+  std::map<RaftServer, SingleVote> mContents;
+};
+
+}
+
+#endif
\ No newline at end of file

test/raft.cc

@@ -31,6 +31,7 @@
 #include "raft/RaftJournal.hh"
 #include "raft/RaftLease.hh"
 #include "raft/RaftContactDetails.hh"
+#include "raft/RaftVoteRegistry.hh"
 #include "Version.hh"
 #include "test-utils.hh"
 #include "RedisParser.hh"
@@ -1076,3 +1077,125 @@ TEST(RaftVoteRequest, Describe) {
   ASSERT_EQ("vote request [candidate=localhost:1234, term=777, lastIndex=999, lastTerm=555]", voteReq.describe(false));
   ASSERT_EQ("pre-vote request [candidate=localhost:1234, term=777, lastIndex=999, lastTerm=555]", voteReq.describe(true));
 }
+
+TEST(RaftVoteRegistry, OneForOneAgainst) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerVote(RaftServer("localhost", 7778), RaftVoteResponse(1, RaftVote::REFUSED));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kElected);
+}
+
+TEST(RaftVoteRegistry, OneForOneVeto) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerVote(RaftServer("localhost", 7778), RaftVoteResponse(1, RaftVote::VETO));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kVetoed);
+}
+
+TEST(RaftVoteRegistry, OneForOneNetErr) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerNetworkError(RaftServer("localhost", 7778));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kElected);
+}
+
+TEST(RaftVoteRegistry, OneForOneParseErr) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerParseError(RaftServer("localhost", 7778));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kElected);
+}
+
+TEST(RaftVoteRegistry, ParsingError) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerParseError(RaftServer("localhost", 7777));
+  registry.registerParseError(RaftServer("localhost", 7778));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kNotElected);
+}
+
+TEST(RaftVoteRegistry, PreVoteParsingError) {
+  RaftVoteRegistry registry(1, true);
+
+  registry.registerParseError(RaftServer("localhost", 7777));
+  registry.registerParseError(RaftServer("localhost", 7778));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kElected);
+}
+
+TEST(RaftVoteRegistry, TwoAgainst) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::REFUSED));
+  registry.registerVote(RaftServer("localhost", 7778), RaftVoteResponse(1, RaftVote::REFUSED));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kNotElected);
+}
+
+TEST(RaftVoteRegistry, TwoVetoes) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::VETO));
+  registry.registerVote(RaftServer("localhost", 7778), RaftVoteResponse(1, RaftVote::VETO));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kVetoed);
+}
+
+TEST(RaftVoteRegistry, TwoFor) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerVote(RaftServer("localhost", 7778), RaftVoteResponse(1, RaftVote::GRANTED));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kElected);
+}
+
+TEST(RaftVoteRegistry, OneFor) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kElected);
+}
+
+TEST(RaftVoteRegistry, OneAgainst) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::REFUSED));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kNotElected);
+}
+
+TEST(RaftVoteRegistry, TwoForOneAgainst) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerVote(RaftServer("localhost", 7778), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerVote(RaftServer("localhost", 7780), RaftVoteResponse(1, RaftVote::REFUSED));
+  registry.registerVote(RaftServer("localhost", 7781), RaftVoteResponse(1, RaftVote::REFUSED));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kElected);
+}
+
+TEST(RaftVoteRegistry, TwoForOneVeto) {
+  RaftVoteRegistry registry(1, false);
+
+  registry.registerVote(RaftServer("localhost", 7777), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerVote(RaftServer("localhost", 7778), RaftVoteResponse(1, RaftVote::GRANTED));
+  registry.registerVote(RaftServer("localhost", 7780), RaftVoteResponse(1, RaftVote::REFUSED));
+  registry.registerVote(RaftServer("localhost", 7781), RaftVoteResponse(1, RaftVote::VETO));
+
+  ASSERT_EQ(registry.determineOutcome(), ElectionOutcome::kVetoed);
+}
+
+
+