Release v1.0.5

stepchowfun · stepchowfun · commit c16504919bd6 · 2023-06-29T04:00:16.000-07:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,7 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [1.0.4] - 2023-06-18
+## [1.0.5] - 2023-06-29
+
+### Changed
+- For efficiency, proposers now use the round from the minimum proposal number returned by the `accept` endpoint to update their own round counter.
+
+## [1.0.4] - 2023-06-29
 
 ### Added
 - Paxos supports a new platform: Windows on AArch64.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "paxos"
-version = "1.0.4"
+version = "1.0.5"
 authors = ["Stephan Boyer <stephan@stephanboyer.com>"]
 edition = "2021"
 description = "An implementation of single-decree Paxos."
diff --git a/README.md b/README.md
@@ -2,22 +2,24 @@
 
 [![Build status](https://github.com/stepchowfun/paxos/workflows/Continuous%20integration/badge.svg?branch=main)](https://github.com/stepchowfun/paxos/actions?query=branch%3Amain)
 
-An reference implementation of single-decree Paxos.
+This is a reference implementation of single-decree Paxos.
 
 ## Configuration
 
 By default, the program looks for a configuration file named `config.yml` in the working directory. This file describes the cluster membership. An [example configuration](https://github.com/stepchowfun/paxos/blob/main/config.yml) is provided in this repository.
 
 ## Usage
 
-For a simple demonstration, run the following commands in separate terminals:
+For a simple demonstration, run the following commands from separate terminals in the repository root:
 
 ```sh
 paxos --node 0 --propose foo
 paxos --node 1 --propose bar
 paxos --node 2 --propose baz
 ```
 
+The cluster will likely achieve consensus immediately after two of the three nodes have been started. The chosen value will be printed to STDOUT by each node in the cluster.
+
 Here are the full usage instructions:
 
 ```
diff --git a/src/acceptor.rs b/src/acceptor.rs
@@ -9,7 +9,7 @@ use {
     serde::{Deserialize, Serialize},
     std::{
         convert::Infallible,
-        io,
+        io::{self, Write},
         net::SocketAddr,
         path::{Path, PathBuf},
         sync::Arc,
@@ -42,7 +42,7 @@ pub struct PrepareResponse {
 // Logic for the "prepare" endpoint
 fn prepare(request: &PrepareRequest, state: &mut State) -> PrepareResponse {
     info!(
-        "Received prepare message:\n{}",
+        "Received prepare request:\n{}",
         serde_yaml::to_string(request).unwrap(), // Serialization is safe.
     );
 
@@ -79,7 +79,7 @@ pub struct AcceptResponse {
 // Logic for the "accept" endpoint
 fn accept(request: &AcceptRequest, state: &mut State) -> AcceptResponse {
     info!(
-        "Received accept message:\n{}",
+        "Received accept request:\n{}",
         serde_yaml::to_string(request).unwrap(), // Serialization is safe.
     );
     if state
@@ -113,9 +113,10 @@ pub struct ChooseResponse;
 
 // Logic for the "choose" endpoint
 fn choose(request: &ChooseRequest, state: &mut State) -> ChooseResponse {
-    info!("Consensus achieved.");
+    info!("Received notification that consensus was achieved.");
     if state.chosen_value.is_none() {
         println!("{}", request.value);
+        io::stdout().flush().unwrap_or(());
         state.chosen_value = Some(request.value.clone());
     }
     ChooseResponse {}
diff --git a/src/main.rs b/src/main.rs
@@ -55,7 +55,6 @@ struct Settings {
 
 // Set up the logger.
 fn set_up_logging() {
-    // Set up the logger.
     Builder::new()
         .filter_module(
             module_path!(),
diff --git a/src/proposer.rs b/src/proposer.rs
@@ -21,6 +21,113 @@ const EXPONENTIAL_BACKOFF_MULTIPLIER: u32 = 2;
 const RESTART_DELAY_MIN: Duration = Duration::from_millis(0);
 const RESTART_DELAY_MAX: Duration = Duration::from_millis(100);
 
+// Propose a value to the cluster.
+pub async fn propose(
+    state: Arc<RwLock<State>>,
+    data_file_path: &Path,
+    nodes: &[SocketAddr],
+    node_index: usize,
+    original_value: &str,
+) -> Result<(), io::Error> {
+    // Retry until the protocol succeeds.
+    loop {
+        // Generate a new proposal number.
+        let proposal_number = {
+            // The `unwrap` is safe since it can only fail if a panic already happened.
+            let mut guard = state.write().await;
+            let proposal_number = generate_proposal_number(nodes, node_index, &mut guard);
+            crate::state::write(&guard, data_file_path).await?;
+            proposal_number
+        };
+
+        // Create an HTTP client.
+        let client = Client::new();
+
+        // Send a prepare message to all the nodes.
+        info!(
+            "Preparing value `{}` with proposal number:\n{}",
+            original_value,
+            // Serialization is safe.
+            serde_yaml::to_string(&proposal_number).unwrap(),
+        );
+        let prepare_responses = broadcast_quorum::<PrepareResponse>(
+            &client,
+            nodes,
+            PREPARE_ENDPOINT,
+            &PrepareRequest { proposal_number },
+        )
+        .await;
+
+        // Determine which value to propose.
+        let new_value = if let Some(accepted_proposal) = prepare_responses
+            .iter()
+            .filter_map(|response| response.accepted_proposal.clone())
+            .max_by_key(|accepted_proposal| accepted_proposal.0)
+        {
+            // There was an accepted proposal. Use that.
+            info!(
+                "Discovered existing value from cluster: {}",
+                accepted_proposal.1,
+            );
+            accepted_proposal.1
+        } else {
+            // Propose the given value.
+            info!("Quorum replied with no existing value.");
+            original_value.to_owned()
+        };
+
+        // Send an accept message to all the nodes.
+        info!(
+            "Requesting acceptance of value `{}` with proposal number:\n{}",
+            new_value,
+            // The `unwrap` is safe because serialization should never fail.
+            serde_yaml::to_string(&proposal_number).unwrap(),
+        );
+        let accept_responses = broadcast_quorum::<AcceptResponse>(
+            &client,
+            nodes,
+            ACCEPT_ENDPOINT,
+            &AcceptRequest {
+                proposal: (proposal_number, new_value.clone()),
+            },
+        )
+        .await;
+
+        // Determine if the proposed value was chosen.
+        let mut value_chosen = true;
+        for response in accept_responses {
+            if response.min_proposal_number > proposal_number {
+                value_chosen = false;
+            }
+
+            // Update the `next_round`, if applicable. The `unwrap` is safe
+            // since it can only fail if a panic already happened.
+            let mut guard = state.write().await;
+            if guard.next_round <= response.min_proposal_number.round {
+                guard.next_round = response.min_proposal_number.round + 1;
+                crate::state::write(&guard, data_file_path).await?;
+            }
+        }
+        if value_chosen {
+            // The protocol succeeded. Notify all the nodes and return.
+            info!("Consensus achieved. Notifying all the nodes.");
+            broadcast_all::<ChooseResponse>(
+                &client,
+                nodes,
+                CHOOSE_ENDPOINT,
+                &ChooseRequest { value: new_value },
+            )
+            .await;
+            info!("All nodes notified.");
+            return Ok(());
+        }
+
+        // The protocol failed. Sleep for a random duration before starting over.
+        info!("Failed to reach consensus. Starting over.");
+        sleep(thread_rng().gen_range(RESTART_DELAY_MIN..RESTART_DELAY_MAX)).await;
+    }
+}
+
 // Generate a new proposal number.
 fn generate_proposal_number(
     nodes: &[SocketAddr],
@@ -124,107 +231,6 @@ async fn broadcast_all<T: DeserializeOwned>(
         .await
 }
 
-// Propose a value to the cluster.
-pub async fn propose(
-    state: Arc<RwLock<State>>,
-    data_file_path: &Path,
-    nodes: &[SocketAddr],
-    node_index: usize,
-    original_value: &str,
-) -> Result<(), io::Error> {
-    // Retry until the protocol succeeds.
-    loop {
-        // Generate a new proposal number.
-        let proposal_number = {
-            // The `unwrap` is safe since it can only fail if a panic already happened.
-            let mut guard = state.write().await;
-            generate_proposal_number(nodes, node_index, &mut guard)
-        };
-
-        // Persist the state.
-        {
-            // The `unwrap` is safe since it can only fail if a panic already happened.
-            let guard = state.read().await;
-            crate::state::write(&guard, data_file_path).await?;
-        }
-
-        // Create an HTTP client.
-        let client = Client::new();
-
-        // Send a prepare message to all the nodes.
-        info!(
-            "Preparing value `{}` with proposal number:\n{}",
-            original_value,
-            // Serialization is safe.
-            serde_yaml::to_string(&proposal_number).unwrap(),
-        );
-        let prepare_responses = broadcast_quorum::<PrepareResponse>(
-            &client,
-            nodes,
-            PREPARE_ENDPOINT,
-            &PrepareRequest { proposal_number },
-        )
-        .await;
-
-        // Determine which value to propose.
-        let new_value = if let Some(accepted_proposal) = prepare_responses
-            .iter()
-            .filter_map(|response| response.accepted_proposal.clone())
-            .max_by_key(|accepted_proposal| accepted_proposal.0)
-        {
-            // There was an accepted proposal. Use that.
-            info!(
-                "Discovered existing value from cluster: {}",
-                accepted_proposal.1,
-            );
-            accepted_proposal.1
-        } else {
-            // Propose the given value.
-            info!("Quorum replied with no existing value.");
-            original_value.to_owned()
-        };
-
-        // Send an accept message to all the nodes.
-        info!(
-            "Requesting acceptance of value `{}` with proposal number:\n{}",
-            new_value,
-            // The `unwrap` is safe because serialization should never fail.
-            serde_yaml::to_string(&proposal_number).unwrap(),
-        );
-        let accept_responses = broadcast_quorum::<AcceptResponse>(
-            &client,
-            nodes,
-            ACCEPT_ENDPOINT,
-            &AcceptRequest {
-                proposal: (proposal_number, new_value.clone()),
-            },
-        )
-        .await;
-
-        // Was the proposed value chosen?
-        if accept_responses
-            .iter()
-            .all(|response| response.min_proposal_number == proposal_number)
-        {
-            // The protocol succeeded. Notify all the nodes and return.
-            info!("Consensus achieved. Notifying all the nodes.");
-            broadcast_all::<ChooseResponse>(
-                &client,
-                nodes,
-                CHOOSE_ENDPOINT,
-                &ChooseRequest { value: new_value },
-            )
-            .await;
-            info!("All nodes notified.");
-            return Ok(());
-        }
-
-        // The protocol failed. Sleep for a random duration before starting over.
-        info!("Failed to reach consensus. Starting over.");
-        sleep(thread_rng().gen_range(RESTART_DELAY_MIN..RESTART_DELAY_MAX)).await;
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use {
