Coordinator recovery WIP.

finnbear · Jun 25, 2023 · c65545d · c65545d
1 parent 82d70ea
commit c65545d
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 9 deletions.
diff --git a/Makefile b/Makefile
@@ -6,4 +6,4 @@ bench:
 
 maelstrom:
 	cargo build --release --features maelstrom --bin maelstrom
-	maelstrom test -w lin-kv --bin target/release/maelstrom --latency 0 --rate 30 --time-limit 30 --concurrency 20 # --nemesis partition --nemesis-interval 10
+	maelstrom test -w lin-kv --bin target/release/maelstrom --latency 0 --rate 30 --time-limit 60 --concurrency 10 # --nemesis partition --nemesis-interval 10
diff --git a/src/bin/maelstrom.rs b/src/bin/maelstrom.rs
@@ -454,7 +454,9 @@ impl Process<LinKv, Wrapper> for KvNode {
                                                         dest: src,
                                                         body: Body::Error(Error {
                                                             in_reply_to: msg_id,
-                                                            text: String::from("read txn conflict"),
+                                                            text: String::from(
+                                                                "write txn conflict",
+                                                            ),
                                                             code: 30,
                                                         }),
                                                     })

diff --git a/src/tapir/client.rs b/src/tapir/client.rs
@@ -7,7 +7,7 @@ use std::{
     future::Future,
     hash::Hash,
     sync::atomic::{AtomicU64, Ordering},
-    time::SystemTime,
+    time::{Duration, SystemTime},
 };
 
 pub struct Client<K: Key, V: Value, T: Transport> {
@@ -77,11 +77,12 @@ impl<K: Key, V: Value, T: Transport<Message = IrMessage<Replica<K, V>>>> Transac
                         continue;
                     }
                 }
+                if matches!(result, OccPrepareResult::TooLate | OccPrepareResult::TooOld) {
+                    continue;
+                }
                 let ok = matches!(result, OccPrepareResult::Ok);
-                use rand::Rng;
                 if inject_fault {
-                    // Induce a coordinator failure.
-                    return None;
+                    T::sleep(Duration::from_secs(u64::MAX / 4)).await;
                 }
                 inner.end(timestamp, ok).await;
 
@@ -95,6 +96,7 @@ impl<K: Key, V: Value, T: Transport<Message = IrMessage<Replica<K, V>>>> Transac
     }
 
     pub fn commit(&self) -> impl Future<Output = Option<Timestamp>> {
-        self.commit_inner(false)
+        use rand::Rng;
+        self.commit_inner(rand::thread_rng().gen_bool(0.02))
     }
 }
diff --git a/src/tapir/replica.rs b/src/tapir/replica.rs
@@ -451,7 +451,7 @@ impl<K: Key, V: Value> IrReplicaUpcalls for Replica<K, V> {
         membership: &IrMembership<T>,
         transport: &T,
     ) {
-        let threshold: u64 = transport.time() - 5 * 1000 * 1000 * 1000;
+        let threshold: u64 = transport.time() - 10 * 1000 * 1000;
         for (transaction_id, (commit, transaction)) in &self.inner.prepared {
             if commit.time > threshold {
                 // Allow the client to finish on its own.

diff --git a/src/tapir/shard_client.rs b/src/tapir/shard_client.rs
@@ -173,7 +173,12 @@ impl<K: Key, V: Value, T: Transport<Message = IrMessage<Replica<K, V>>>> ShardTr
                         OccPrepareResult::Fail => {
                             return CR::Prepare(OccPrepareResult::Fail);
                         }
-                        OccPrepareResult::TooLate | OccPrepareResult::TooOld => unimplemented!(),
+                        OccPrepareResult::TooLate => {
+                            return CR::Prepare(OccPrepareResult::TooLate);
+                        }
+                        OccPrepareResult::TooOld => {
+                            return CR::Prepare(OccPrepareResult::TooOld);
+                        }
                     }
                 }