V1 Stable after PLDI submission

Author: Farzin

.gitignore

@@ -43,3 +43,5 @@ build/
 
 # Custom makefile artifacts
 .deps/
+
+wellcoordination/workload/**

2.log

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+PERCENTAGES="5 15 25";
+
+for n in $( seq 3 7 ); do
+        for p in $PERCENTAGES; do
+		grep -r -w "total average" wellcoordination/workload/$n-4000000-$p/cart/results/* | sort -t: -n -k2 | awk '{split($0,a); count += 1; sum += a[8]} END {print sum/count;}';
+	done
+done

3.log

@@ -1,7 +1,7 @@
 source load_modules.sh
-./build.py crash-consensus;
+./build.py crash-consensus -b debug;
 crash-consensus/libgen/export.sh gcc-release;
-crash-consensus/demo/using_conan_fully/build.sh gcc-release;
+crash-consensus/demo/using_conan_fully/build.sh gcc-debug;
 #crash-consensus/experiments/build.sh
 
 # fix this

cart-res.sh~

@@ -1,10 +1,6 @@
 #include "consensus.hpp"
 
 #include <iostream>
-// #include <algorithm>
-// #include <functional>
-
-// #include "protocol/nb-protocol.cpp"
 
 namespace dory {
 RdmaConsensus::RdmaConsensus(int my_id, std::vector<int>& remote_ids,
@@ -20,7 +16,7 @@ RdmaConsensus::RdmaConsensus(int my_id, std::vector<int>& remote_ids,
       LOGGER_INIT(logger, ConsensusConfig::logger_prefix) {
   using namespace units;
 
-  allocated_size = 2_GiB;
+  allocated_size = 4_GiB;
   alignment = 64;
 
   run();

compile.sh~

@@ -55,6 +55,9 @@ class RdmaConsensus {
   int propose(uint8_t *buf, size_t len);
 
   inline int potentialLeader() { return potential_leader; }
+  inline bool amILeader() { return am_I_leader.load(); }
+
+  void stopHeartbeatThread() {leader_election->stopHeartbreat();}
 
   inline std::pair<uint64_t, uint64_t> proposedReplicatedRange() {
     return std::make_pair(majW->range_start, majW->range_end);

crash-consensus/src/consensus.cpp

@@ -40,9 +40,12 @@ ProposeError Consensus::propose(uint8_t *buf, size_t len) {
 }
 
 int Consensus::potentialLeader() { return impl->potentialLeader(); }
+bool Consensus::amILeader() { return impl->amILeader(); }
 bool Consensus::blockedResponse() { return impl->response_blocked->load(); }
 
 std::pair<uint64_t, uint64_t> Consensus::proposedReplicatedRange() {
   return impl->proposedReplicatedRange();
 }
+
+void Consensus::stopHeartbeatThread() { impl->stopHeartbeatThread(); }
 }  // namespace dory

crash-consensus/src/consensus.hpp

@@ -37,9 +37,12 @@ class Consensus {
 
   ProposeError propose(uint8_t *buf, size_t len);
   int potentialLeader();
+  bool amILeader();
   bool blockedResponse();
   std::pair<uint64_t, uint64_t> proposedReplicatedRange();
 
+  void stopHeartbeatThread();
+
  private:
   std::unique_ptr<RdmaConsensus> impl;
 };

crash-consensus/src/crash-consensus.cpp

@@ -101,7 +101,6 @@ class Follower {
 
  private:
   void run() {
-    std::cout << "started follower thread" << std::endl;
     int loops = 0;
     constexpr unsigned mask = (1 << 14) - 1;  // Must be power of 2 minus 1
 
@@ -123,17 +122,15 @@ class Follower {
           log_mutex.lock();
         }
       }
-      std::cout << "checking queue" << std::endl;
       auto has_next = iter->sampleNext();
       if (!has_next) {
         continue;
       }
 
       ParsedSlot pslot(iter->location());
       // std::cout << "Discovered element on position " <<
-      // uintptr_t(iter->location()) << std::endl; std::cout << "Accepted
-      // proposal " << pslot.acceptedProposal()
-      //           << std::endl;
+      // uintptr_t(iter->location()) << std::endl; 
+      // std::cout << "Accepted proposal " << pslot.acceptedProposal() << std::endl;
       // std::cout << "First undecided offset " << pslot.firstUndecidedOffset()
       //           << std::endl;
       // auto [buf, len] = pslot.payload();

crash-consensus/src/crash-consensus.hpp

@@ -44,6 +44,8 @@ class LeaderHeartbeat {
     // Careful, there is a move assignment happening!
   }
 
+  std::atomic<int> start_id;
+
   void startPoller() {
     read_seq = 0;
 
@@ -202,7 +204,7 @@ class LeaderHeartbeat {
     //   std::cout << std::endl;
     // }
 
-    if (leader_pid() == ctx->cc.my_id) {
+    if (leader_pid(start_id.load()) == ctx->cc.my_id) {
       want_leader.store(true);
     } else {
       std::this_thread::sleep_for(std::chrono::milliseconds(50));
@@ -242,13 +244,15 @@ class LeaderHeartbeat {
   };
 
  private:
-  int leader_pid() {
+  int leader_pid(int start_index) {
     int leader_id = -1;
 
-    for (auto &pid : ids) {
+    for (int i = 0; i < static_cast<int>(ids.size()); i++) {
+      int pid = ids[i];
       // std::cout << pid << " " << status[pid].consecutive_updates <<
       // std::endl;
       if (status[pid].consecutive_updates > 2) {
+        // leader_id = pid;
         leader_id = pid;
         break;
       }
@@ -533,7 +537,10 @@ class LeaderSwitcher {
         want_leader{&heartbeat->wantLeaderSignal()},
         read_slots{ctx->scratchpad.writeLeaderChangeSlots()},
         sz{read_slots.size()},
+        // leader_start_index{(heartbeat->start_id == 0 ? 0 : heartbeat->start_id + 1)},
         permission_asker{ctx} {
+    leader_start_index.store((heartbeat->start_id.load() == 0 ? 0 : heartbeat->start_id.load() + 1));
+    std::cout << "inside leaderswitcher constructor: " << leader_start_index << std::endl;
     prepareScanner();
   }
 
@@ -545,7 +552,7 @@ class LeaderSwitcher {
     int force_reset = 0;
     auto constexpr shift = 8 * sizeof(uintptr_t) - 1;
 
-    for (int i = 0; i < static_cast<int>(sz); i++) {
+    for (int i = leader_start_index.load(); i < static_cast<int>(sz); i++) {
       reading[i] = *reinterpret_cast<uint64_t *>(read_slots[i]);
       force_reset = static_cast<int>(reading[i] >> shift);
       reading[i] &= (1UL << shift) - 1;
@@ -556,12 +563,12 @@ class LeaderSwitcher {
         break;
       }
     }
-
     // If you discovered a new request for a leader, notify the main event loop
     // to give permissions to him and switch to follower.
     if (requester > 0) {
-      // std::cout << "Process with pid " << requester
-      //           << " asked for permissions" << std::endl;
+      std::cout << "index: " << leader_start_index.load() << std::endl;
+      std::cout << "Process with pid " << requester
+                << " asked for permissions" << std::endl;
       leader.store(dory::Leader(requester, reading[requester], force_reset));
       want_leader->store(false);
     } else {
@@ -594,10 +601,10 @@ class LeaderSwitcher {
     Leader current_leader = leader.load();
     if (current_leader != prev_leader || force_permission_request) {
       // std::cout << "Adjusting connections to leader ("
-      //           << int(current_leader.requester) << " "
-      //           << current_leader.requester_value << ") " << (current_leader
-      //           != prev_leader) << " " << force_permission_request <<
-      //           std::endl;
+                // << int(current_leader.requester) << " "
+                // << current_leader.requester_value << ") " << (current_leader
+                // != prev_leader) << " " << force_permission_request <<
+                // std::endl;
 
       auto orig_leader = prev_leader;
       prev_leader = current_leader;
@@ -611,13 +618,13 @@ class LeaderSwitcher {
 
           // GET_TIMESTAMP(ts_start);
 
-          // std::cout << "Asking for permissions: " << hard_reset << std::endl;
+          std::cout << "Asking for permissions: " << hard_reset << std::endl;
           // Ask for permission. Wait for everybody to reply
           permission_asker.askForPermissions(hard_reset);
 
           // GET_TIMESTAMP(ts_mid);
 
-          // std::cout << "Waiting for approval" << std::endl;
+          std::cout << "Waiting for approval" << std::endl;
           // In order to avoid a distributed deadlock (when two processes try
           // to become leaders at the same time), we bail whe the leader
           // changes.
@@ -643,7 +650,7 @@ class LeaderSwitcher {
           // std::cout << "Asking for permissions: " << hard_reset << std::endl;
 
           // std::cout << "I (process " << c_ctx->my_id << ") got leader "
-          //           << "approval" << std::endl;
+                    // << "approval" << std::endl;
 
           // GET_TIMESTAMP(ts_start);
           if (hard_reset) {
@@ -828,6 +835,7 @@ class LeaderSwitcher {
   std::vector<uint8_t *> dummy;
   std::vector<uint8_t *> &read_slots;
   size_t sz;
+  std::atomic<int> leader_start_index; 
 
   LeaderPermissionAsker permission_asker;
 
@@ -873,6 +881,15 @@ class LeaderElection {
     return leader_switcher.leaderSignal();
   }
 
+  // made public
+  void stopHeartbreat() {
+    if (hb_started) {
+      hb_exit_signal.set_value();
+      heartbeat_thd.join();
+      hb_started = false;
+    }
+  }
+
  private:
   void startHeartbeat() {
     if (hb_started) {
@@ -881,7 +898,15 @@ class LeaderElection {
     hb_started = true;
 
     leader_heartbeat = LeaderHeartbeat(&ctx);
+    if(threadConfig.prefix == "Secondary-")
+      leader_heartbeat.start_id.store(1);
+    else
+      leader_heartbeat.start_id.store(0);
     std::future<void> ftr = hb_exit_signal.get_future();
+    uint64_t start_hb = std::chrono::duration_cast<std::chrono::microseconds>(
+            std::chrono::high_resolution_clock::now().time_since_epoch())
+            .count();
+    std::cout << "started heartbeat at " << start_hb << std::endl;
     heartbeat_thd = std::thread([this, ftr = std::move(ftr)]() {
       leader_heartbeat.startPoller();
 
@@ -947,14 +972,22 @@ class LeaderElection {
           response_blocked.store(false);
           leader_heartbeat.scanHeartbeats();
         } else if (prev_command == 'c') {
-	  response_blocked.store(true);
+	        response_blocked.store(true);
           leader_heartbeat.retract();
         }
 
         prev_command = current_command;
 
         std::this_thread::sleep_for(std::chrono::milliseconds(1));
 
+        // ADDED FARZIN
+        // std::cout << "checking exit" << std::endl;
+          if (ftr.wait_for(std::chrono::seconds(0)) ==
+              std::future_status::ready) {
+            std::cout << "exiting heartbeat..." << std::endl;
+            break;
+          }
+
         // if (i == 0) {
         //   if (ftr.wait_for(std::chrono::seconds(0)) !=
         //       std::future_status::timeout) {
@@ -965,7 +998,8 @@ class LeaderElection {
         // std::this_thread::sleep_for(std::chrono::seconds(10));
       }
 
-      file_watcher_thd.join();
+      // file_watcher_thd.join();
+      file_watcher_thd.detach();
     });
 
     if (threadConfig.pinThreads) {
@@ -977,32 +1011,25 @@ class LeaderElection {
     }
   }
 
-  void stopHeartbreat() {
-    if (hb_started) {
-      hb_exit_signal.set_value();
-      heartbeat_thd.join();
-      hb_started = false;
-    }
-  }
-
   void startLeaderSwitcher() {
     if (switcher_started) {
       throw std::runtime_error("Already started");
     }
     switcher_started = true;
-
     leader_switcher = LeaderSwitcher(&ctx, &leader_heartbeat);
     std::future<void> ftr = switcher_exit_signal.get_future();
     switcher_thd = std::thread([this, ftr = std::move(ftr)]() {
       leader_switcher.startPoller();
       for (unsigned long long i = 0;; i = (i + 1) & iterations_ftr_check) {
         leader_switcher.scanPermissions();
-        if (i == 0) {
-          if (ftr.wait_for(std::chrono::seconds(0)) !=
-              std::future_status::timeout) {
-            break;
-          }
-        }
+        // if (i == 0) {
+        //   std::cout << "inside exit check" << std::endl;
+        //   if (ftr.wait_for(std::chrono::seconds(0)) !=
+        //       std::future_status::timeout) {
+        //     std::cout << "exiting heartbeat..." << std::endl;
+        //     break;
+        //   }
+        // }
       }
     });
 

crash-consensus/src/follower.hpp

@@ -25,7 +25,7 @@
 using namespace dory;
 using dory::units::operator""_GiB;
 
-size_t allocated_size = 1_GiB;
+size_t allocated_size = 4_GiB;
 int alignment = 64;
 
 

crash-consensus/src/leader-switch.hpp

@@ -0,0 +1,25 @@
+#!/bin/bash
+
+PERCENTAGES="25 15 5";
+
+for p in $PERCENTAGES; do
+	printf $p" :\n";
+       for n in $( seq 3 7 ); do
+    #    printf $n" :\n";
+		grep -r -w "sum_response_time:" wellcoordination/workload/$n-4000000-$p/$1/results/* | sort -t: -n -k2 | head -1 | awk '{split($0,a); print 4000000000/a[2]}';
+
+	done
+done
+
+#grep -r -w "total average" wellcoordination/workload/4-4000000-50/courseware/results/hamsaz* | sort -t: -n -k2 | awk '{split($0,a); sum += a[8]; count += 1;} END{print sum/count}'
+# for p in $PERCENTAGES; do
+# 	printf "prec: "$p" "; 
+#         for n in $( seq 3 7 ); do
+# 		printf "node: "$n" \n";
+# 		for i in $( seq 0 4 ) ; do
+# 			grep -r -w "calls to $i" wellcoordination/workload/4-4000000-$p/$1/results/$2* | sort -t: -n -k2 | awk '{split($0,a); if(a[9] != "-nan") {sum += a[9]; count += 1;}} END{print sum/count;}'
+# 		done
+# 		printf "###\n";
+# 	done
+# 	printf "****\n";
+# done