From 4b04bc8612a39ad7ff5ccf27db155ccc61179567 Mon Sep 17 00:00:00 2001 From: Michael Vines Date: Sat, 16 Mar 2019 08:31:19 -0700 Subject: [PATCH] Move thin_client RPC requests into rpc_request; de-mut thin_client --- bench-tps/src/bench.rs | 15 +- bench-tps/src/main.rs | 8 +- client/src/rpc_request.rs | 206 +++++++++++++++++- client/src/thin_client.rs | 270 ++++-------------------- core/src/cluster_tests.rs | 16 +- core/src/local_cluster.rs | 21 +- core/src/replicator.rs | 6 +- core/src/storage_stage.rs | 2 +- proposals/src/cluster-test-framework.md | 4 +- tests/thin_client.rs | 12 +- 10 files changed, 285 insertions(+), 275 deletions(-) diff --git a/bench-tps/src/bench.rs b/bench-tps/src/bench.rs index 8b6c1c4b3..98c8955f8 100644 --- a/bench-tps/src/bench.rs +++ b/bench-tps/src/bench.rs @@ -52,7 +52,7 @@ pub fn sample_tx_count( v: &ContactInfo, sample_period: u64, ) { - let mut client = create_client(v.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(v.client_facing_addr(), FULLNODE_PORT_RANGE); let mut now = Instant::now(); let mut initial_tx_count = client.transaction_count(); let mut max_tps = 0.0; @@ -182,7 +182,7 @@ pub fn generate_txs( reclaim: bool, contact_info: &ContactInfo, ) { - let mut client = create_client(contact_info.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(contact_info.client_facing_addr(), FULLNODE_PORT_RANGE); let blockhash = client.get_recent_blockhash(); let tx_count = source.len(); println!("Signing transactions... {} (reclaim={})", tx_count, reclaim); @@ -292,7 +292,7 @@ pub fn do_tx_transfers( } } -pub fn verify_funding_transfer(client: &mut ThinClient, tx: &Transaction, amount: u64) -> bool { +pub fn verify_funding_transfer(client: &ThinClient, tx: &Transaction, amount: u64) -> bool { for a in &tx.account_keys[1..] { if client.get_balance(a).unwrap_or(0) >= amount { return true; @@ -305,7 +305,7 @@ pub fn verify_funding_transfer(client: &mut ThinClient, tx: &Transaction, amount /// fund the dests keys by spending all of the source keys into MAX_SPENDS_PER_TX /// on every iteration. This allows us to replay the transfers because the source is either empty, /// or full -pub fn fund_keys(client: &mut ThinClient, source: &Keypair, dests: &[Keypair], lamports: u64) { +pub fn fund_keys(client: &ThinClient, source: &Keypair, dests: &[Keypair], lamports: u64) { let total = lamports * dests.len() as u64; let mut funded: Vec<(&Keypair, u64)> = vec![(source, total)]; let mut notfunded: Vec<&Keypair> = dests.iter().collect(); @@ -398,12 +398,7 @@ pub fn fund_keys(client: &mut ThinClient, source: &Keypair, dests: &[Keypair], l } } -pub fn airdrop_lamports( - client: &mut ThinClient, - drone_addr: &SocketAddr, - id: &Keypair, - tx_count: u64, -) { +pub fn airdrop_lamports(client: &ThinClient, drone_addr: &SocketAddr, id: &Keypair, tx_count: u64) { let starting_balance = client.poll_get_balance(&id.pubkey()).unwrap_or(0); metrics_submit_lamport_balance(starting_balance); println!("starting balance {}", starting_balance); diff --git a/bench-tps/src/main.rs b/bench-tps/src/main.rs index 96f1cd593..85654fe15 100644 --- a/bench-tps/src/main.rs +++ b/bench-tps/src/main.rs @@ -63,7 +63,7 @@ fn main() { } let cluster_entrypoint = nodes[0].clone(); // Pick the first node, why not? - let mut client = create_client(cluster_entrypoint.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(cluster_entrypoint.client_facing_addr(), FULLNODE_PORT_RANGE); let mut barrier_client = create_client(cluster_entrypoint.client_facing_addr(), FULLNODE_PORT_RANGE); @@ -94,13 +94,13 @@ fn main() { if num_lamports_per_account > keypair0_balance { let extra = num_lamports_per_account - keypair0_balance; let total = extra * (gen_keypairs.len() as u64); - airdrop_lamports(&mut client, &drone_addr, &id, total); + airdrop_lamports(&client, &drone_addr, &id, total); println!("adding more lamports {}", extra); - fund_keys(&mut client, &id, &gen_keypairs, extra); + fund_keys(&client, &id, &gen_keypairs, extra); } let start = gen_keypairs.len() - (tx_count * 2) as usize; let keypairs = &gen_keypairs[start..]; - airdrop_lamports(&mut barrier_client, &drone_addr, &barrier_source_keypair, 1); + airdrop_lamports(&barrier_client, &drone_addr, &barrier_source_keypair, 1); println!("Get last ID..."); let mut blockhash = client.get_recent_blockhash(); diff --git a/client/src/rpc_request.rs b/client/src/rpc_request.rs index 6e36537cc..f1518258e 100644 --- a/client/src/rpc_request.rs +++ b/client/src/rpc_request.rs @@ -1,15 +1,19 @@ +use bs58; use log::*; use reqwest; use reqwest::header::CONTENT_TYPE; use serde_json::{json, Value}; +use solana_sdk::account::Account; +use solana_sdk::hash::Hash; +use solana_sdk::pubkey::Pubkey; +use solana_sdk::signature::Signature; use solana_sdk::timing::{DEFAULT_TICKS_PER_SLOT, NUM_TICKS_PER_SECOND}; +use std::io; use std::net::SocketAddr; use std::thread::sleep; -use std::time::Duration; +use std::time::{Duration, Instant}; use std::{error, fmt}; -use solana_sdk::pubkey::Pubkey; - #[derive(Clone)] pub struct RpcClient { pub client: reqwest::Client, @@ -49,6 +53,202 @@ impl RpcClient { Ok(res) } + pub fn get_account_data(&self, pubkey: &Pubkey) -> io::Result>> { + let params = json!([format!("{}", pubkey)]); + let response = self.make_rpc_request(RpcRequest::GetAccountInfo, Some(params)); + match response { + Ok(account_json) => { + let account: Account = + serde_json::from_value(account_json).expect("deserialize account"); + Ok(Some(account.data)) + } + Err(error) => { + debug!("get_account_data failed: {:?}", error); + Err(io::Error::new( + io::ErrorKind::Other, + "get_account_data failed", + )) + } + } + } + + /// Request the balance of the user holding `pubkey`. This method blocks + /// until the server sends a response. If the response packet is dropped + /// by the network, this method will hang indefinitely. + pub fn get_balance(&self, pubkey: &Pubkey) -> io::Result { + let params = json!([format!("{}", pubkey)]); + let response = self.make_rpc_request(RpcRequest::GetAccountInfo, Some(params)); + + response + .and_then(|account_json| { + let account: Account = + serde_json::from_value(account_json).expect("deserialize account"); + trace!("Response account {:?} {:?}", pubkey, account); + trace!("get_balance {:?}", account.lamports); + Ok(account.lamports) + }) + .map_err(|error| { + debug!("Response account {}: None (error: {:?})", pubkey, error); + io::Error::new(io::ErrorKind::Other, "AccountNotFound") + }) + } + + /// Request the transaction count. If the response packet is dropped by the network, + /// this method will try again 5 times. + pub fn transaction_count(&self) -> u64 { + debug!("transaction_count"); + for _tries in 0..5 { + let response = self.make_rpc_request(RpcRequest::GetTransactionCount, None); + + match response { + Ok(value) => { + debug!("transaction_count response: {:?}", value); + let transaction_count = value.as_u64().unwrap(); + return transaction_count; + } + Err(error) => { + debug!("transaction_count failed: {:?}", error); + } + }; + } + 0 + } + + /// Request the last Entry ID from the server without blocking. + /// Returns the blockhash Hash or None if there was no response from the server. + pub fn try_get_recent_blockhash(&self, mut num_retries: u64) -> Option { + loop { + let response = self.make_rpc_request(RpcRequest::GetRecentBlockhash, None); + + match response { + Ok(value) => { + let blockhash_str = value.as_str().unwrap(); + let blockhash_vec = bs58::decode(blockhash_str).into_vec().unwrap(); + return Some(Hash::new(&blockhash_vec)); + } + Err(error) => { + debug!("thin_client get_recent_blockhash error: {:?}", error); + num_retries -= 1; + if num_retries == 0 { + return None; + } + } + } + } + } + + /// Request the last Entry ID from the server. This method blocks + /// until the server sends a response. + pub fn get_recent_blockhash(&self) -> Hash { + loop { + if let Some(hash) = self.try_get_recent_blockhash(10) { + return hash; + } + } + } + + /// Request a new last Entry ID from the server. This method blocks + /// until the server sends a response. + pub fn get_next_blockhash(&self, previous_blockhash: &Hash) -> Hash { + self.get_next_blockhash_ext(previous_blockhash, &|| { + sleep(Duration::from_millis(100)); + }) + } + + fn get_next_blockhash_ext(&self, previous_blockhash: &Hash, func: &Fn()) -> Hash { + loop { + let blockhash = self.get_recent_blockhash(); + if blockhash != *previous_blockhash { + break blockhash; + } + debug!("Got same blockhash ({:?}), will retry...", blockhash); + func() + } + } + + pub fn poll_balance_with_timeout( + &self, + pubkey: &Pubkey, + polling_frequency: &Duration, + timeout: &Duration, + ) -> io::Result { + let now = Instant::now(); + loop { + match self.get_balance(&pubkey) { + Ok(bal) => { + return Ok(bal); + } + Err(e) => { + sleep(*polling_frequency); + if now.elapsed() > *timeout { + return Err(e); + } + } + }; + } + } + + pub fn poll_get_balance(&self, pubkey: &Pubkey) -> io::Result { + self.poll_balance_with_timeout(pubkey, &Duration::from_millis(100), &Duration::from_secs(1)) + } + + /// Poll the server to confirm a transaction. + pub fn poll_for_signature(&self, signature: &Signature) -> io::Result<()> { + let now = Instant::now(); + while !self.check_signature(signature) { + if now.elapsed().as_secs() > 15 { + // TODO: Return a better error. + return Err(io::Error::new(io::ErrorKind::Other, "signature not found")); + } + sleep(Duration::from_millis(250)); + } + Ok(()) + } + + /// Check a signature in the bank. This method blocks + /// until the server sends a response. + pub fn check_signature(&self, signature: &Signature) -> bool { + trace!("check_signature: {:?}", signature); + let params = json!([format!("{}", signature)]); + + loop { + let response = + self.make_rpc_request(RpcRequest::ConfirmTransaction, Some(params.clone())); + + match response { + Ok(confirmation) => { + let signature_status = confirmation.as_bool().unwrap(); + if signature_status { + trace!("Response found signature"); + } else { + trace!("Response signature not found"); + } + + return signature_status; + } + Err(err) => { + debug!("check_signature request failed: {:?}", err); + } + }; + } + } + pub fn fullnode_exit(&self) -> io::Result { + let response = self + .make_rpc_request(RpcRequest::FullnodeExit, None) + .map_err(|err| { + io::Error::new( + io::ErrorKind::Other, + format!("FullnodeExit request failure: {:?}", err), + ) + })?; + serde_json::from_value(response).map_err(|err| { + io::Error::new( + io::ErrorKind::Other, + format!("FullnodeExit parse failure: {:?}", err), + ) + }) + } + pub fn retry_make_rpc_request( &self, request: &RpcRequest, diff --git a/client/src/thin_client.rs b/client/src/thin_client.rs index 4ca6f35b1..74a657db1 100644 --- a/client/src/thin_client.rs +++ b/client/src/thin_client.rs @@ -3,14 +3,11 @@ //! messages to the network directly. The binary encoding of its messages are //! unstable and may change in future releases. -use crate::rpc_request::{RpcClient, RpcRequest, RpcRequestHandler}; +use crate::rpc_request::RpcClient; use bincode::serialize_into; -use bs58; use log::*; -use serde_json::json; use solana_metrics; use solana_metrics::influxdb; -use solana_sdk::account::Account; use solana_sdk::hash::Hash; use solana_sdk::packet::PACKET_DATA_SIZE; use solana_sdk::pubkey::Pubkey; @@ -21,13 +18,10 @@ use solana_sdk::transaction::Transaction; use std; use std::io; use std::net::{SocketAddr, UdpSocket}; -use std::thread::sleep; -use std::time::Duration; -use std::time::Instant; +use std::time::{Duration, Instant}; /// An object for querying and sending transactions to the network. pub struct ThinClient { - rpc_addr: SocketAddr, transactions_addr: SocketAddr, transactions_socket: UdpSocket, rpc_client: RpcClient, @@ -42,7 +36,6 @@ impl ThinClient { transactions_socket: UdpSocket, ) -> Self { Self::new_from_client( - rpc_addr, transactions_addr, transactions_socket, RpcClient::new_socket(rpc_addr), @@ -56,18 +49,16 @@ impl ThinClient { timeout: Duration, ) -> Self { let rpc_client = RpcClient::new_socket_with_timeout(rpc_addr, timeout); - Self::new_from_client(rpc_addr, transactions_addr, transactions_socket, rpc_client) + Self::new_from_client(transactions_addr, transactions_socket, rpc_client) } fn new_from_client( - rpc_addr: SocketAddr, transactions_addr: SocketAddr, transactions_socket: UdpSocket, rpc_client: RpcClient, ) -> Self { ThinClient { rpc_client, - rpc_addr, transactions_addr, transactions_socket, } @@ -88,7 +79,7 @@ impl ThinClient { /// Retry a sending a signed Transaction to the server for processing. pub fn retry_transfer( - &mut self, + &self, keypair: &Keypair, transaction: &mut Transaction, tries: usize, @@ -142,237 +133,66 @@ impl ThinClient { result } - pub fn get_account_data(&mut self, pubkey: &Pubkey) -> io::Result>> { - let params = json!([format!("{}", pubkey)]); - let response = self - .rpc_client - .make_rpc_request(RpcRequest::GetAccountInfo, Some(params)); - match response { - Ok(account_json) => { - let account: Account = - serde_json::from_value(account_json).expect("deserialize account"); - Ok(Some(account.data)) - } - Err(error) => { - debug!("get_account_data failed: {:?}", error); - Err(io::Error::new( - io::ErrorKind::Other, - "get_account_data failed", - )) - } - } + pub fn get_account_data(&self, pubkey: &Pubkey) -> io::Result>> { + self.rpc_client.get_account_data(pubkey) } - /// Request the balance of the user holding `pubkey`. This method blocks - /// until the server sends a response. If the response packet is dropped - /// by the network, this method will hang indefinitely. - pub fn get_balance(&mut self, pubkey: &Pubkey) -> io::Result { - trace!("get_balance sending request to {}", self.rpc_addr); - let params = json!([format!("{}", pubkey)]); - let response = self - .rpc_client - .make_rpc_request(RpcRequest::GetAccountInfo, Some(params)); - - response - .and_then(|account_json| { - let account: Account = - serde_json::from_value(account_json).expect("deserialize account"); - trace!("Response account {:?} {:?}", pubkey, account); - trace!("get_balance {:?}", account.lamports); - Ok(account.lamports) - }) - .map_err(|error| { - debug!("Response account {}: None (error: {:?})", pubkey, error); - io::Error::new(io::ErrorKind::Other, "AccountNotFound") - }) + pub fn get_balance(&self, pubkey: &Pubkey) -> io::Result { + self.rpc_client.get_balance(pubkey) } - /// Request the transaction count. If the response packet is dropped by the network, - /// this method will try again 5 times. - pub fn transaction_count(&mut self) -> u64 { - debug!("transaction_count"); - for _tries in 0..5 { - let response = self - .rpc_client - .make_rpc_request(RpcRequest::GetTransactionCount, None); - - match response { - Ok(value) => { - debug!("transaction_count response: {:?}", value); - let transaction_count = value.as_u64().unwrap(); - return transaction_count; - } - Err(error) => { - debug!("transaction_count failed: {:?}", error); - } - }; - } - 0 + pub fn transaction_count(&self) -> u64 { + self.rpc_client.transaction_count() } - /// Request the last Entry ID from the server without blocking. - /// Returns the blockhash Hash or None if there was no response from the server. - pub fn try_get_recent_blockhash(&mut self, mut num_retries: u64) -> Option { - loop { - trace!("try_get_recent_blockhash send_to {}", &self.rpc_addr); - let response = self - .rpc_client - .make_rpc_request(RpcRequest::GetRecentBlockhash, None); - - match response { - Ok(value) => { - let blockhash_str = value.as_str().unwrap(); - let blockhash_vec = bs58::decode(blockhash_str).into_vec().unwrap(); - return Some(Hash::new(&blockhash_vec)); - } - Err(error) => { - debug!("thin_client get_recent_blockhash error: {:?}", error); - num_retries -= 1; - if num_retries == 0 { - return None; - } - } - } - } + pub fn try_get_recent_blockhash(&self, num_retries: u64) -> Option { + self.rpc_client.try_get_recent_blockhash(num_retries) } - /// Request the last Entry ID from the server. This method blocks - /// until the server sends a response. - pub fn get_recent_blockhash(&mut self) -> Hash { - loop { - trace!("get_recent_blockhash send_to {}", &self.rpc_addr); - if let Some(hash) = self.try_get_recent_blockhash(10) { - return hash; - } - } + pub fn get_recent_blockhash(&self) -> Hash { + self.rpc_client.get_recent_blockhash() } - /// Request a new last Entry ID from the server. This method blocks - /// until the server sends a response. - pub fn get_next_blockhash(&mut self, previous_blockhash: &Hash) -> Hash { - self.get_next_blockhash_ext(previous_blockhash, &|| { - sleep(Duration::from_millis(100)); - }) - } - pub fn get_next_blockhash_ext(&mut self, previous_blockhash: &Hash, func: &Fn()) -> Hash { - loop { - let blockhash = self.get_recent_blockhash(); - if blockhash != *previous_blockhash { - break blockhash; - } - debug!("Got same blockhash ({:?}), will retry...", blockhash); - func() - } - } - - pub fn submit_poll_balance_metrics(elapsed: &Duration) { - solana_metrics::submit( - influxdb::Point::new("thinclient") - .add_tag("op", influxdb::Value::String("get_balance".to_string())) - .add_field( - "duration_ms", - influxdb::Value::Integer(timing::duration_as_ms(elapsed) as i64), - ) - .to_owned(), - ); + pub fn get_next_blockhash(&self, previous_blockhash: &Hash) -> Hash { + self.rpc_client.get_next_blockhash(previous_blockhash) } pub fn poll_balance_with_timeout( - &mut self, + &self, pubkey: &Pubkey, polling_frequency: &Duration, timeout: &Duration, ) -> io::Result { + self.rpc_client + .poll_balance_with_timeout(pubkey, polling_frequency, timeout) + } + + pub fn poll_get_balance(&self, pubkey: &Pubkey) -> io::Result { + self.rpc_client.poll_get_balance(pubkey) + } + + pub fn poll_for_signature(&self, signature: &Signature) -> io::Result<()> { + self.rpc_client.poll_for_signature(signature) + } + + pub fn check_signature(&self, signature: &Signature) -> bool { let now = Instant::now(); - loop { - match self.get_balance(&pubkey) { - Ok(bal) => { - ThinClient::submit_poll_balance_metrics(&now.elapsed()); - return Ok(bal); - } - Err(e) => { - sleep(*polling_frequency); - if now.elapsed() > *timeout { - ThinClient::submit_poll_balance_metrics(&now.elapsed()); - return Err(e); - } - } - }; - } + let result = self.rpc_client.check_signature(signature); + + solana_metrics::submit( + influxdb::Point::new("thinclient") + .add_tag("op", influxdb::Value::String("check_signature".to_string())) + .add_field( + "duration_ms", + influxdb::Value::Integer(timing::duration_as_ms(&now.elapsed()) as i64), + ) + .to_owned(), + ); + result } - pub fn poll_get_balance(&mut self, pubkey: &Pubkey) -> io::Result { - self.poll_balance_with_timeout(pubkey, &Duration::from_millis(100), &Duration::from_secs(1)) - } - - /// Poll the server to confirm a transaction. - pub fn poll_for_signature(&mut self, signature: &Signature) -> io::Result<()> { - let now = Instant::now(); - while !self.check_signature(signature) { - if now.elapsed().as_secs() > 15 { - // TODO: Return a better error. - return Err(io::Error::new(io::ErrorKind::Other, "signature not found")); - } - sleep(Duration::from_millis(250)); - } - Ok(()) - } - - /// Check a signature in the bank. This method blocks - /// until the server sends a response. - pub fn check_signature(&mut self, signature: &Signature) -> bool { - trace!("check_signature: {:?}", signature); - let params = json!([format!("{}", signature)]); - let now = Instant::now(); - - loop { - let response = self - .rpc_client - .make_rpc_request(RpcRequest::ConfirmTransaction, Some(params.clone())); - - match response { - Ok(confirmation) => { - let signature_status = confirmation.as_bool().unwrap(); - if signature_status { - trace!("Response found signature"); - } else { - trace!("Response signature not found"); - } - solana_metrics::submit( - influxdb::Point::new("thinclient") - .add_tag("op", influxdb::Value::String("check_signature".to_string())) - .add_field( - "duration_ms", - influxdb::Value::Integer( - timing::duration_as_ms(&now.elapsed()) as i64 - ), - ) - .to_owned(), - ); - return signature_status; - } - Err(err) => { - debug!("check_signature request failed: {:?}", err); - } - }; - } - } - pub fn fullnode_exit(&mut self) -> io::Result { - trace!("fullnode_exit sending request to {}", self.rpc_addr); - let response = self - .rpc_client - .make_rpc_request(RpcRequest::FullnodeExit, None) - .map_err(|error| { - debug!("Response from {} fullndoe_exit: {}", self.rpc_addr, error); - io::Error::new(io::ErrorKind::Other, "FullodeExit request failure") - })?; - serde_json::from_value(response).map_err(|error| { - debug!( - "ParseError: from {} fullndoe_exit: {}", - self.rpc_addr, error - ); - io::Error::new(io::ErrorKind::Other, "FullodeExit parse failure") - }) + pub fn fullnode_exit(&self) -> io::Result { + self.rpc_client.fullnode_exit() } } @@ -383,7 +203,7 @@ impl Drop for ThinClient { } pub fn retry_get_balance( - client: &mut ThinClient, + client: &ThinClient, bob_pubkey: &Pubkey, expected_balance: Option, ) -> Option { diff --git a/core/src/cluster_tests.rs b/core/src/cluster_tests.rs index e0b20a174..66f65f1bf 100644 --- a/core/src/cluster_tests.rs +++ b/core/src/cluster_tests.rs @@ -28,7 +28,7 @@ pub fn spend_and_verify_all_nodes( assert!(cluster_nodes.len() >= nodes); for ingress_node in &cluster_nodes { let random_keypair = Keypair::new(); - let mut client = create_client(ingress_node.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(ingress_node.client_facing_addr(), FULLNODE_PORT_RANGE); let bal = client .poll_get_balance(&funding_keypair.pubkey()) .expect("balance in source"); @@ -44,14 +44,14 @@ pub fn spend_and_verify_all_nodes( .retry_transfer(&funding_keypair, &mut transaction, 5) .unwrap(); for validator in &cluster_nodes { - let mut client = create_client(validator.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(validator.client_facing_addr(), FULLNODE_PORT_RANGE); client.poll_for_signature(&sig).unwrap(); } } } pub fn send_many_transactions(node: &ContactInfo, funding_keypair: &Keypair, num_txs: u64) { - let mut client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); for _ in 0..num_txs { let random_keypair = Keypair::new(); let bal = client @@ -75,12 +75,12 @@ pub fn fullnode_exit(entry_point_info: &ContactInfo, nodes: usize) { let cluster_nodes = discover(&entry_point_info.gossip, nodes).unwrap(); assert!(cluster_nodes.len() >= nodes); for node in &cluster_nodes { - let mut client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); assert!(client.fullnode_exit().unwrap()); } sleep(Duration::from_millis(SLOT_MILLIS)); for node in &cluster_nodes { - let mut client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); assert!(client.fullnode_exit().is_err()); } } @@ -126,7 +126,7 @@ pub fn kill_entry_and_spend_and_verify_rest( solana_logger::setup(); let cluster_nodes = discover(&entry_point_info.gossip, nodes).unwrap(); assert!(cluster_nodes.len() >= nodes); - let mut client = create_client(entry_point_info.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(entry_point_info.client_facing_addr(), FULLNODE_PORT_RANGE); info!("sleeping for an epoch"); sleep(Duration::from_millis(SLOT_MILLIS * DEFAULT_SLOTS_PER_EPOCH)); info!("done sleeping for an epoch"); @@ -140,7 +140,7 @@ pub fn kill_entry_and_spend_and_verify_rest( continue; } - let mut client = create_client(ingress_node.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(ingress_node.client_facing_addr(), FULLNODE_PORT_RANGE); let bal = client .poll_get_balance(&funding_keypair.pubkey()) .expect("balance in source"); @@ -195,7 +195,7 @@ fn poll_all_nodes_for_signature( if validator.id == entry_point_info.id { continue; } - let mut client = create_client(validator.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(validator.client_facing_addr(), FULLNODE_PORT_RANGE); client.poll_for_signature(&sig)?; } diff --git a/core/src/local_cluster.rs b/core/src/local_cluster.rs index 1c16e141f..6d03f67a1 100644 --- a/core/src/local_cluster.rs +++ b/core/src/local_cluster.rs @@ -115,7 +115,7 @@ impl LocalCluster { } fn add_validator(&mut self, fullnode_config: &FullnodeConfig, stake: u64) { - let mut client = create_client( + let client = create_client( self.entry_point_info.client_facing_addr(), FULLNODE_PORT_RANGE, ); @@ -131,19 +131,14 @@ impl LocalCluster { // Send each validator some lamports to vote let validator_balance = - Self::transfer(&mut client, &self.funding_keypair, &validator_pubkey, stake); + Self::transfer(&client, &self.funding_keypair, &validator_pubkey, stake); info!( "validator {} balance {}", validator_pubkey, validator_balance ); - Self::create_and_fund_vote_account( - &mut client, - &voting_keypair, - &validator_keypair, - stake - 1, - ) - .unwrap(); + Self::create_and_fund_vote_account(&client, &voting_keypair, &validator_keypair, stake - 1) + .unwrap(); let validator_server = Fullnode::new( validator_node, @@ -160,13 +155,13 @@ impl LocalCluster { fn add_replicator(&mut self) { let replicator_keypair = Arc::new(Keypair::new()); - let mut client = create_client( + let client = create_client( self.entry_point_info.client_facing_addr(), FULLNODE_PORT_RANGE, ); Self::transfer( - &mut client, + &client, &self.funding_keypair, &replicator_keypair.pubkey(), 1, @@ -196,7 +191,7 @@ impl LocalCluster { } fn transfer( - client: &mut ThinClient, + client: &ThinClient, source_keypair: &Keypair, dest_pubkey: &Pubkey, lamports: u64, @@ -218,7 +213,7 @@ impl LocalCluster { } fn create_and_fund_vote_account( - client: &mut ThinClient, + client: &ThinClient, vote_account: &Keypair, from_account: &Arc, amount: u64, diff --git a/core/src/replicator.rs b/core/src/replicator.rs index 488da8afd..11800976f 100644 --- a/core/src/replicator.rs +++ b/core/src/replicator.rs @@ -310,11 +310,11 @@ impl Replicator { } fn submit_mining_proof(&self) { - let mut client = create_client( + let client = create_client( self.cluster_entrypoint.client_facing_addr(), FULLNODE_PORT_RANGE, ); - Self::get_airdrop_lamports(&mut client, &self.keypair, &self.cluster_entrypoint); + Self::get_airdrop_lamports(&client, &self.keypair, &self.cluster_entrypoint); let blockhash = client.get_recent_blockhash(); let mut tx = StorageTransaction::new_mining_proof( @@ -378,7 +378,7 @@ impl Replicator { } fn get_airdrop_lamports( - client: &mut ThinClient, + client: &ThinClient, keypair: &Keypair, cluster_entrypoint: &ContactInfo, ) { diff --git a/core/src/storage_stage.rs b/core/src/storage_stage.rs index f4d765a0d..ab9ae9b85 100644 --- a/core/src/storage_stage.rs +++ b/core/src/storage_stage.rs @@ -228,7 +228,7 @@ impl StorageStage { account_to_create: Option, ) -> io::Result<()> { let contact_info = cluster_info.read().unwrap().my_data(); - let mut client = create_client_with_timeout( + let client = create_client_with_timeout( contact_info.client_facing_addr(), FULLNODE_PORT_RANGE, Duration::from_secs(5), diff --git a/proposals/src/cluster-test-framework.md b/proposals/src/cluster-test-framework.md index e0d59fff5..1cb9fad3c 100644 --- a/proposals/src/cluster-test-framework.md +++ b/proposals/src/cluster-test-framework.md @@ -102,7 +102,7 @@ pub fn test_large_invalid_gossip_nodes( let cluster = discover(&entry_point_info, num_nodes); // Poison the cluster. - let mut client = create_client(entry_point_info.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(entry_point_info.client_facing_addr(), FULLNODE_PORT_RANGE); for _ in 0..(num_nodes * 100) { client.gossip_push( cluster_info::invalid_contact_info() @@ -112,7 +112,7 @@ pub fn test_large_invalid_gossip_nodes( // Force refresh of the active set. for node in &cluster { - let mut client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(node.client_facing_addr(), FULLNODE_PORT_RANGE); client.gossip_refresh_active_set(); } diff --git a/tests/thin_client.rs b/tests/thin_client.rs index 7e87ef0fe..aa87303eb 100644 --- a/tests/thin_client.rs +++ b/tests/thin_client.rs @@ -23,7 +23,7 @@ fn test_thin_client_basic() { let bob_pubkey = Keypair::new().pubkey(); discover(&leader_data.gossip, 1).unwrap(); - let mut client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); let transaction_count = client.transaction_count(); assert_eq!(transaction_count, 0); @@ -54,7 +54,7 @@ fn test_bad_sig() { let bob_pubkey = Keypair::new().pubkey(); discover(&leader_data.gossip, 1).unwrap(); - let mut client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); let blockhash = client.get_recent_blockhash(); @@ -85,7 +85,7 @@ fn test_register_vote_account() { let (server, leader_data, alice, ledger_path) = new_fullnode_for_tests(); discover(&leader_data.gossip, 1).unwrap(); - let mut client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); // Create the validator account, transfer some lamports to that account let validator_keypair = Keypair::new(); @@ -106,7 +106,7 @@ fn test_register_vote_account() { let signature = client.transfer_signed(&transaction).unwrap(); client.poll_for_signature(&signature).unwrap(); - let balance = retry_get_balance(&mut client, &vote_account_id, Some(1)) + let balance = retry_get_balance(&client, &vote_account_id, Some(1)) .expect("Expected balance for new account to exist"); assert_eq!(balance, 1); @@ -139,7 +139,7 @@ fn test_transaction_count() { solana_logger::setup(); let addr = "0.0.0.0:1234".parse().unwrap(); let transactions_socket = UdpSocket::bind("0.0.0.0:0").unwrap(); - let mut client = ThinClient::new_socket_with_timeout( + let client = ThinClient::new_socket_with_timeout( addr, addr, transactions_socket, @@ -155,7 +155,7 @@ fn test_zero_balance_after_nonzero() { let bob_keypair = Keypair::new(); discover(&leader_data.gossip, 1).unwrap(); - let mut client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); + let client = create_client(leader_data.client_facing_addr(), FULLNODE_PORT_RANGE); let blockhash = client.get_recent_blockhash(); info!("test_thin_client blockhash: {:?}", blockhash);