Make the logging less verbose

Author: stepchowfun

.gitignore

@@ -1,3 +1,4 @@
 # [tag:gitignore] Keep this in sync with [ref:excluded_input_paths].
+.data
 /artifacts/
 /target/

src/acceptor.rs

@@ -41,7 +41,7 @@ pub struct PrepareResponse {
 
 // Logic for the "prepare" endpoint
 fn prepare(request: &PrepareRequest, state: &mut State) -> PrepareResponse {
-    info!(
+    debug!(
         "Received prepare request:\n{}",
         serde_yaml::to_string(request).unwrap(), // Serialization is safe.
     );
@@ -78,7 +78,7 @@ pub struct AcceptResponse {
 
 // Logic for the "accept" endpoint
 fn accept(request: &AcceptRequest, state: &mut State) -> AcceptResponse {
-    info!(
+    debug!(
         "Received accept request:\n{}",
         serde_yaml::to_string(request).unwrap(), // Serialization is safe.
     );
@@ -113,8 +113,8 @@ pub struct ChooseResponse;
 
 // Logic for the "choose" endpoint
 fn choose(request: &ChooseRequest, state: &mut State) -> ChooseResponse {
-    info!("Received notification that consensus was achieved.");
     if state.chosen_value.is_none() {
+        info!("Consensus was achieved.");
         println!("{}", request.value);
         io::stdout().flush().unwrap_or(());
         state.chosen_value = Some(request.value.clone());

src/main.rs

@@ -3,6 +3,7 @@
 mod acceptor;
 mod config;
 mod proposer;
+mod rpc;
 mod state;
 
 #[macro_use]
@@ -228,7 +229,7 @@ async fn settings() -> io::Result<Settings> {
     );
 
     // Determine the data file path [tag:data_file_path_has_parent].
-    let data_file_path = Path::join(data_dir_path, format!("{}:{}", ip, port));
+    let data_file_path = Path::join(data_dir_path, format!("{}-{}", ip, port));
 
     // Return the settings.
     Ok(Settings {
@@ -279,22 +280,26 @@ async fn main() {
         }
     }
 
-    // Run the acceptor and the proposer, if applicable.
+    // Run the acceptor and the proposer, if applicable. Also, broadcasr any chosen value from the
+    // persisted state.
     if let Err(error) = try_join!(
         acceptor::acceptor(state.clone(), &settings.data_file_path, settings.address),
         async {
-            if let Some(value) = &settings.proposal {
-                propose(
-                    state,
-                    &settings.data_file_path,
-                    &settings.nodes,
-                    settings.node_index,
-                    value,
-                )
-                .await
-            } else {
-                Ok(())
+            // If there's a value to propose and no chosen value in the persisted state, propose the
+            // value.
+            if state.read().await.chosen_value.is_none() {
+                if let Some(value) = &settings.proposal {
+                    propose(
+                        state.clone(),
+                        &settings.data_file_path,
+                        &settings.nodes,
+                        settings.node_index,
+                        value,
+                    )
+                    .await?;
+                }
             }
+            Ok(())
         },
     ) {
         error!("{}", error);

src/proposer.rs

@@ -4,23 +4,33 @@ use {
             AcceptRequest, AcceptResponse, ChooseRequest, ChooseResponse, PrepareRequest,
             PrepareResponse, ACCEPT_ENDPOINT, CHOOSE_ENDPOINT, PREPARE_ENDPOINT,
         },
+        rpc::{broadcast_all, broadcast_quorum},
         state::{ProposalNumber, State},
     },
-    futures::{stream::FuturesUnordered, StreamExt},
-    hyper::{client::HttpConnector, Body, Client, Method, Request},
+    hyper::Client,
     rand::{thread_rng, Rng},
-    serde::{de::DeserializeOwned, Serialize},
-    std::{cmp::min, io, net::SocketAddr, path::Path, sync::Arc, time::Duration},
+    std::{io, net::SocketAddr, path::Path, sync::Arc, time::Duration},
     tokio::{sync::RwLock, time::sleep},
 };
 
 // Duration constants
-const EXPONENTIAL_BACKOFF_MIN: Duration = Duration::from_millis(100);
-const EXPONENTIAL_BACKOFF_MAX: Duration = Duration::from_secs(2);
-const EXPONENTIAL_BACKOFF_MULTIPLIER: u32 = 2;
 const RESTART_DELAY_MIN: Duration = Duration::from_millis(0);
 const RESTART_DELAY_MAX: Duration = Duration::from_millis(100);
 
+// Generate a new proposal number.
+fn generate_proposal_number(
+    nodes: &[SocketAddr],
+    node_index: usize,
+    state: &mut State,
+) -> ProposalNumber {
+    let proposal_number = ProposalNumber {
+        round: state.next_round,
+        proposer_address: nodes[node_index],
+    };
+    state.next_round += 1;
+    proposal_number
+}
+
 // Propose a value to the cluster.
 pub async fn propose(
     state: Arc<RwLock<State>>,
@@ -31,6 +41,9 @@ pub async fn propose(
 ) -> Result<(), io::Error> {
     // Retry until the protocol succeeds.
     loop {
+        // Create an HTTP client.
+        let client = Client::new();
+
         // Generate a new proposal number.
         let proposal_number = {
             // The `unwrap` is safe since it can only fail if a panic already happened.
@@ -40,11 +53,8 @@ pub async fn propose(
             proposal_number
         };
 
-        // Create an HTTP client.
-        let client = Client::new();
-
         // Send a prepare message to all the nodes.
-        info!(
+        debug!(
             "Preparing value `{}` with proposal number:\n{}",
             original_value,
             // Serialization is safe.
@@ -65,19 +75,19 @@ pub async fn propose(
             .max_by_key(|accepted_proposal| accepted_proposal.0)
         {
             // There was an accepted proposal. Use that.
-            info!(
+            debug!(
                 "Discovered existing value from cluster: {}",
                 accepted_proposal.1,
             );
             accepted_proposal.1
         } else {
             // Propose the given value.
-            info!("Quorum replied with no existing value.");
+            debug!("Quorum replied with no existing value.");
             original_value.to_owned()
         };
 
         // Send an accept message to all the nodes.
-        info!(
+        debug!(
             "Requesting acceptance of value `{}` with proposal number:\n{}",
             new_value,
             // The `unwrap` is safe because serialization should never fail.
@@ -110,127 +120,24 @@ pub async fn propose(
         }
         if value_chosen {
             // The protocol succeeded. Notify all the nodes and return.
-            info!("Consensus achieved. Notifying all the nodes.");
+            debug!("Consensus achieved. Notifying all the nodes.");
             broadcast_all::<ChooseResponse>(
                 &client,
                 nodes,
                 CHOOSE_ENDPOINT,
                 &ChooseRequest { value: new_value },
             )
             .await;
-            info!("All nodes notified.");
+            info!("Proposer finished.");
             return Ok(());
         }
 
         // The protocol failed. Sleep for a random duration before starting over.
-        info!("Failed to reach consensus. Starting over.");
+        debug!("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],
-    node_index: usize,
-    state: &mut State,
-) -> ProposalNumber {
-    let proposal_number = ProposalNumber {
-        round: state.next_round,
-        proposer_address: nodes[node_index],
-    };
-    state.next_round += 1;
-    proposal_number
-}
-
-// Send a request without retries.
-async fn try_to_send<T: DeserializeOwned>(
-    client: &Client<HttpConnector, Body>,
-    node: SocketAddr,
-    endpoint: &str,
-    payload: &impl Serialize,
-) -> Result<T, hyper::Error> {
-    Ok(bincode::deserialize(
-        &hyper::body::to_bytes(
-            client
-                .request(
-                    Request::builder()
-                        .method(Method::POST)
-                        .uri(format!("http://{}{}", node, endpoint))
-                        // The `unwrap` is safe because serialization should never fail.
-                        .body(Body::from(bincode::serialize(&payload).unwrap()))
-                        .unwrap(), // Safe since we constructed a well-formed request
-                )
-                .await?
-                .into_body(),
-        )
-        .await?,
-    )
-    .unwrap()) // Safe under non-Byzantine conditions
-}
-
-// Send a request, retrying with exponential backoff until it succeeds.
-async fn send<T: DeserializeOwned>(
-    client: &Client<HttpConnector, Body>,
-    node: SocketAddr,
-    endpoint: &str,
-    payload: &impl Serialize,
-) -> T {
-    // The delay between requests
-    let mut delay = EXPONENTIAL_BACKOFF_MIN;
-
-    // Retry until the request succeeds.
-    loop {
-        // Send the request.
-        match try_to_send(client, node, endpoint, payload).await {
-            Ok(response) => {
-                return response;
-            }
-            Err(error) => {
-                // Log the error.
-                error!("Received error: {}", error);
-            }
-        }
-
-        // Sleep before retrying.
-        sleep(delay).await;
-        delay = min(
-            delay * EXPONENTIAL_BACKOFF_MULTIPLIER,
-            EXPONENTIAL_BACKOFF_MAX,
-        );
-    }
-}
-
-// Send a request to all nodes. Return once a majority of responses come in.
-async fn broadcast_quorum<T: DeserializeOwned>(
-    client: &Client<HttpConnector, Body>,
-    nodes: &[SocketAddr],
-    endpoint: &str,
-    payload: &impl Serialize,
-) -> Vec<T> {
-    nodes
-        .iter()
-        .map(|node| send(client, *node, endpoint, payload))
-        .collect::<FuturesUnordered<_>>()
-        .take(nodes.len() / 2 + 1)
-        .collect()
-        .await
-}
-
-// Send a request to all nodes. Return once all responses come in.
-async fn broadcast_all<T: DeserializeOwned>(
-    client: &Client<HttpConnector, Body>,
-    nodes: &[SocketAddr],
-    endpoint: &str,
-    payload: &impl Serialize,
-) -> Vec<T> {
-    nodes
-        .iter()
-        .map(|node| send(client, *node, endpoint, payload))
-        .collect::<FuturesUnordered<_>>()
-        .collect()
-        .await
-}
-
 #[cfg(test)]
 mod tests {
     use {