Merge pull request #102 from commerceblock/develop

Release 0.4.9 - State sync improvement

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "coordinator"
-version = "0.4.8"
+version = "0.4.9"
 authors = ["nkostoulas <[email protected]>"]
 description = "Guardnode Coordinator implementation for the Commerceblock Covalence system"
 homepage = "https://github.com/commerceblock"

scripts/api_request.sh

@@ -1,31 +1,22 @@
 #!/bin/bash
 
-NUM=1   # set $1 for load testing
-if [ ! -z "$1" ]; then
-    NUM=$1
-    echo "a"
-fi
-
-for ((i=1;i<=$NUM;i++)); do
-    RESP=$(curl -s -S -X POST -H "Content-Type: application/json"\
-        -d "{\"jsonrpc\": \"2.0\", \"method\": \"getrequests\", \"params\" : {}, \"id\":1 }"\
-        userApi:passwordApi@localhost:3333)
-    echo $i
-    if [ $i == 1 ]; then
-        echo $RESP | jq -r '.result' | jq -r .
-    fi
-done
+echo "Getting all requests..."
+RESP=$(curl -s -S -X POST -H "Content-Type: application/json"\
+    -d "{\"jsonrpc\": \"2.0\", \"method\": \"getrequests\", \"params\" : {}, \"id\":1 }" -u $1 $2)
+echo $RESP | jq -r '.result' | jq -r .
 
 TXID=$(echo $RESP | jq -r ".result" | jq -r  ".requests[0].request.txid")
+if [ ! -z $3 ]; then
+    TXID=$3
+fi
 
+echo "Getting request $TXID..."
 RESP=$(curl -s -S -X POST -H "Content-Type: application/json"\
-    -d "{\"jsonrpc\": \"2.0\", \"method\": \"getrequestresponses\", \"params\" : {\"txid\": \"$TXID\"}, \"id\":2 }"\
-    userApi:passwordApi@localhost:3333)
+    -d "{\"jsonrpc\": \"2.0\", \"method\": \"getrequestresponse\", \"params\" : {\"txid\": \"$TXID\"}, \"id\":2 }" -u $1 $2)
 
 echo $RESP | jq -r '.'
 
 RESP=$(curl -s -S -X POST -H "Content-Type: application/json"\
-    -d "{\"jsonrpc\": \"2.0\", \"method\": \"getrequest\", \"params\" : {\"txid\": \"$TXID\"}, \"id\":3 }"\
-    userApi:passwordApi@localhost:3333)
+    -d "{\"jsonrpc\": \"2.0\", \"method\": \"getrequest\", \"params\" : {\"txid\": \"$TXID\"}, \"id\":3 }"  -u $1 $2)
 
 echo $RESP | jq -r '.'

src/challenger.rs

@@ -5,7 +5,7 @@
 
 use std::collections::HashSet;
 use std::sync::mpsc::{Receiver, RecvTimeoutError};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, RwLock};
 use std::{thread, time};
 
 use bitcoin::hashes::sha256d;
@@ -94,15 +94,15 @@ fn get_challenge_response(
 pub fn run_challenge_request<T: Service, K: ClientChain, D: Storage>(
     service: &T,
     clientchain: &K,
-    challenge_state: Arc<Mutex<Option<ChallengeState>>>,
+    challenge_state: Arc<RwLock<Option<ChallengeState>>>,
     verify_rx: &Receiver<ChallengeResponse>,
     storage: Arc<D>,
     verify_duration: time::Duration,
     challenge_duration: time::Duration,
     challenge_frequency: u64,
     refresh_delay: time::Duration,
 ) -> Result<()> {
-    let request = challenge_state.lock().unwrap().as_ref().unwrap().request.clone(); // clone as const and drop mutex
+    let request = challenge_state.read().unwrap().as_ref().unwrap().request.clone(); // clone as const and drop mutex
     let mut response = storage.get_response(request.txid)?.unwrap_or(Response::new());
     info! {"Running challenge request: {:?}", request.txid};
     let mut prev_challenge_height: u64 = 0;
@@ -119,10 +119,10 @@ pub fn run_challenge_request<T: Service, K: ClientChain, D: Storage>(
 
         info! {"sending challenge..."}
         let challenge_hash = clientchain.send_challenge()?;
-        challenge_state.lock().unwrap().as_mut().unwrap().latest_challenge = Some(challenge_hash);
+        challenge_state.write().unwrap().as_mut().unwrap().latest_challenge = Some(challenge_hash);
 
         if let Err(e) = verify_challenge(&challenge_hash, clientchain, verify_duration) {
-            challenge_state.lock().unwrap().as_mut().unwrap().latest_challenge = None; // stop receiving responses
+            challenge_state.write().unwrap().as_mut().unwrap().latest_challenge = None; // stop receiving responses
             return Err(e);
         }
 
@@ -133,7 +133,7 @@ pub fn run_challenge_request<T: Service, K: ClientChain, D: Storage>(
             challenge_duration,
         )?);
         storage.save_response(request.txid, &response)?;
-        challenge_state.lock().unwrap().as_mut().unwrap().latest_challenge = None; // stop receiving responses
+        challenge_state.write().unwrap().as_mut().unwrap().latest_challenge = None; // stop receiving responses
         prev_challenge_height = challenge_height; // update prev height
     }
     info! {"Challenge request ended"}
@@ -650,7 +650,7 @@ mod tests {
         let res = run_challenge_request(
             &service,
             &clientchain,
-            Arc::new(Mutex::new(Some(challenge_state.clone()))),
+            Arc::new(RwLock::new(Some(challenge_state.clone()))),
             &vrx,
             storage.clone(),
             time::Duration::from_millis(10),
@@ -684,7 +684,7 @@ mod tests {
         let res = run_challenge_request(
             &service,
             &clientchain,
-            Arc::new(Mutex::new(Some(challenge_state.clone()))),
+            Arc::new(RwLock::new(Some(challenge_state.clone()))),
             &vrx,
             storage.clone(),
             time::Duration::from_millis(10),
@@ -726,7 +726,7 @@ mod tests {
         assert!(run_challenge_request(
             &service,
             &clientchain,
-            Arc::new(Mutex::new(Some(challenge_state))),
+            Arc::new(RwLock::new(Some(challenge_state))),
             &vrx,
             storage.clone(),
             time::Duration::from_millis(10),
@@ -745,7 +745,7 @@ mod tests {
         assert!(run_challenge_request(
             &service,
             &clientchain,
-            Arc::new(Mutex::new(Some(challenge_state))),
+            Arc::new(RwLock::new(Some(challenge_state))),
             &vrx,
             storage.clone(),
             time::Duration::from_millis(10),
@@ -765,7 +765,7 @@ mod tests {
         assert!(run_challenge_request(
             &service,
             &clientchain,
-            Arc::new(Mutex::new(Some(challenge_state))),
+            Arc::new(RwLock::new(Some(challenge_state))),
             &vrx,
             Arc::new(storage_err),
             time::Duration::from_millis(10),
@@ -787,7 +787,7 @@ mod tests {
         let res = run_challenge_request(
             &service,
             &clientchain,
-            Arc::new(Mutex::new(Some(challenge_state))),
+            Arc::new(RwLock::new(Some(challenge_state))),
             &vrx,
             storage.clone(),
             time::Duration::from_millis(10),
@@ -815,7 +815,7 @@ mod tests {
         let res = run_challenge_request(
             &service,
             &clientchain,
-            Arc::new(Mutex::new(Some(challenge_state))),
+            Arc::new(RwLock::new(Some(challenge_state))),
             &vrx,
             storage.clone(),
             time::Duration::from_millis(10),

src/coordinator.rs

@@ -3,7 +3,7 @@
 //! Coordinator entry point for spawning all components
 
 use std::sync::mpsc::{channel, Receiver, Sender};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, RwLock};
 use std::{thread, time};
 
 use bitcoin::hashes::{hex::FromHex, sha256d};
@@ -30,7 +30,7 @@ pub fn run(config: Config) -> Result<()> {
 
     // create a challenge state mutex to share between challenger and listener.
     // initially None
-    let shared_challenge = Arc::new(Mutex::new(None));
+    let shared_challenge = Arc::new(RwLock::new(None));
     // and a channel for sending responses from listener to challenger
     let (verify_tx, verify_rx): (Sender<ChallengeResponse>, Receiver<ChallengeResponse>) = channel();
     // start listener along with a oneshot channel to send shutdown message
@@ -57,7 +57,7 @@ pub fn run(config: Config) -> Result<()> {
                     info! {"{}", serde_json::to_string_pretty(&resp).unwrap()};
                 }
                 // Reset challenge state to None.
-                *shared_challenge.lock().unwrap() = None;
+                *shared_challenge.write().unwrap() = None;
 
                 info! {"Sleeping for {} sec...", config.block_time}
                 thread::sleep(time::Duration::from_secs(config.block_time))
@@ -86,7 +86,7 @@ pub fn run_request<T: Service, K: ClientChain, D: Storage>(
     service: &T,
     clientchain: &K,
     storage: Arc<D>,
-    shared_challenge: Arc<Mutex<Option<ChallengeState>>>,
+    shared_challenge: Arc<RwLock<Option<ChallengeState>>>,
     verify_rx: &Receiver<ChallengeResponse>,
     genesis_hash: sha256d::Hash,
 ) -> Result<Option<sha256d::Hash>> {
@@ -105,7 +105,7 @@ pub fn run_request<T: Service, K: ClientChain, D: Storage>(
             )?;
 
             // modify challenge state for the new challenge request
-            *shared_challenge.lock().unwrap() = Some(challenge);
+            *shared_challenge.write().unwrap() = Some(challenge);
 
             // run challenge request storing expected responses
             match ::challenger::run_challenge_request(
@@ -121,7 +121,7 @@ pub fn run_request<T: Service, K: ClientChain, D: Storage>(
             ) {
                 Ok(()) => {
                     // update end clientchain height with final height
-                    let mut shared_ch_lock = shared_challenge.lock().unwrap();
+                    let mut shared_ch_lock = shared_challenge.write().unwrap();
                     let ch_final = shared_ch_lock.as_mut().unwrap();
                     ch_final.request.end_blockheight_clientchain = clientchain.get_blockheight()?;
                     info!(

src/listener.rs

@@ -5,7 +5,7 @@
 use std::net::ToSocketAddrs;
 use std::str::FromStr;
 use std::sync::mpsc::Sender;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, RwLock};
 use std::thread;
 
 use bitcoin::consensus::serialize;
@@ -71,7 +71,7 @@ impl ChallengeProof {
 /// challenger to receive
 fn handle_challengeproof(
     req: Request<Body>,
-    challenge: Arc<Mutex<Option<ChallengeState>>>,
+    challenge: Arc<RwLock<Option<ChallengeState>>>,
     challenge_resp: Sender<ChallengeResponse>,
 ) -> impl Future<Item = Response<Body>, Error = hyper::Error> + Send {
     let resp = req.into_body().concat2().map(move |body| {
@@ -82,7 +82,7 @@ fn handle_challengeproof(
                 // parse challenge proof from json
                 Ok(proof) => {
                     // check for an active challenge
-                    let ch_lock = challenge.lock().unwrap();
+                    let ch_lock = challenge.read().unwrap();
                     if let Some(ch) = ch_lock.as_ref() {
                         if let Some(h) = ch.latest_challenge {
                             // check challenge proof bid exists
@@ -123,7 +123,7 @@ fn handle_challengeproof(
 /// and to the /challengeproof POST uri for receiving challenges from guardnodes
 fn handle(
     req: Request<Body>,
-    challenge: Arc<Mutex<Option<ChallengeState>>>,
+    challenge: Arc<RwLock<Option<ChallengeState>>>,
     challenge_resp: Sender<ChallengeResponse>,
 ) -> impl Future<Item = Response<Body>, Error = hyper::Error> + Send {
     let resp = match (req.method(), req.uri().path()) {
@@ -156,7 +156,7 @@ fn response(status: StatusCode, message: String) -> Response<Body> {
 /// of the coordinator
 pub fn run_listener(
     listener_host: &String,
-    challenge: Arc<Mutex<Option<ChallengeState>>>,
+    challenge: Arc<RwLock<Option<ChallengeState>>>,
     ch_resp: Sender<ChallengeResponse>,
 ) -> Handle {
     let addr: Vec<_> = listener_host
@@ -309,7 +309,7 @@ mod tests {
         let _challenge_state = gen_challenge_state_with_challenge(&gen_dummy_hash(3), &chl_hash);
         let bid_txid = _challenge_state.bids.iter().next().unwrap().txid;
         let bid_pubkey = _challenge_state.bids.iter().next().unwrap().pubkey;
-        let challenge_state = Arc::new(Mutex::new(Some(_challenge_state)));
+        let challenge_state = Arc::new(RwLock::new(Some(_challenge_state)));
 
         // Request get /
         let data = "";
@@ -449,7 +449,7 @@ mod tests {
         let _challenge_state = gen_challenge_state_with_challenge(&gen_dummy_hash(1), &chl_hash);
         let bid_txid = _challenge_state.bids.iter().next().unwrap().txid;
         let bid_pubkey = _challenge_state.bids.iter().next().unwrap().pubkey;
-        let challenge_state = Arc::new(Mutex::new(Some(_challenge_state)));
+        let challenge_state = Arc::new(RwLock::new(Some(_challenge_state)));
 
         // Request body data empty
         let data = "";
@@ -528,7 +528,7 @@ mod tests {
         assert!(resp_rx.try_recv() == Err(TryRecvError::Empty)); // check receiver empty
 
         // No active challenge (hash is None) so request rejected
-        challenge_state.lock().unwrap().as_mut().unwrap().latest_challenge = None;
+        challenge_state.write().unwrap().as_mut().unwrap().latest_challenge = None;
         let data = r#"
         {
             "txid": "0000000000000000000000000000000000000000000000000000000000000000",
@@ -548,7 +548,7 @@ mod tests {
                     .wait()
             })
             .wait();
-        challenge_state.lock().unwrap().as_mut().unwrap().latest_challenge = Some(chl_hash);
+        challenge_state.write().unwrap().as_mut().unwrap().latest_challenge = Some(chl_hash);
         assert!(resp_rx.try_recv() == Err(TryRecvError::Empty)); // check receiver empty
 
         // Invalid bid on request body (txid does not exist)