From 95cc36af96d12767f9ecbc95b457587337fd5e34 Mon Sep 17 00:00:00 2001 From: Tyera Eulberg Date: Fri, 5 Apr 2019 21:57:59 -0600 Subject: [PATCH] Impl SyncClient and AsyncClient for ThinClient --- bench-tps/src/bench.rs | 1 + client/src/thin_client.rs | 159 +++++++++++++++++++++++++++++--------- core/src/local_cluster.rs | 3 +- core/src/storage_stage.rs | 1 + 4 files changed, 125 insertions(+), 39 deletions(-) diff --git a/bench-tps/src/bench.rs b/bench-tps/src/bench.rs index 40b2200a1..ac401ec35 100644 --- a/bench-tps/src/bench.rs +++ b/bench-tps/src/bench.rs @@ -13,6 +13,7 @@ use solana_metrics::influxdb; use solana_sdk::hash::Hash; use solana_sdk::pubkey::Pubkey; use solana_sdk::signature::{Keypair, KeypairUtil}; +use solana_sdk::sync_client::SyncClient; use solana_sdk::system_instruction; use solana_sdk::system_transaction; use solana_sdk::timing::timestamp; diff --git a/client/src/thin_client.rs b/client/src/thin_client.rs index 864e77368..2645a52e4 100644 --- a/client/src/thin_client.rs +++ b/client/src/thin_client.rs @@ -6,11 +6,17 @@ use crate::rpc_client::RpcClient; use bincode::{serialize_into, serialized_size}; use log::*; +use solana_sdk::async_client::AsyncClient; use solana_sdk::hash::Hash; +use solana_sdk::instruction::Instruction; +use solana_sdk::message::Message; use solana_sdk::packet::PACKET_DATA_SIZE; use solana_sdk::pubkey::Pubkey; -use solana_sdk::signature::{Keypair, Signature}; -use solana_sdk::transaction::Transaction; +use solana_sdk::signature::{Keypair, KeypairUtil, Signature}; +use solana_sdk::sync_client::SyncClient; +use solana_sdk::system_instruction; +use solana_sdk::transaction::{self, Transaction}; +use solana_sdk::transport::Result as TransportResult; use std::error; use std::io; use std::net::{SocketAddr, UdpSocket}; @@ -80,9 +86,29 @@ impl ThinClient { transaction: &mut Transaction, tries: usize, min_confirmed_blocks: usize, + ) -> io::Result { + self.send_and_confirm_transaction(&[keypair], transaction, tries, min_confirmed_blocks) + } + + /// Retry sending a signed Transaction with one signing Keypair to the server for processing. + pub fn retry_transfer( + &self, + keypair: &Keypair, + transaction: &mut Transaction, + tries: usize, + ) -> io::Result { + self.send_and_confirm_transaction(&[keypair], transaction, tries, 0) + } + + /// Retry sending a signed Transaction to the server for processing + pub fn send_and_confirm_transaction( + &self, + keypairs: &[&Keypair], + transaction: &mut Transaction, + tries: usize, + min_confirmed_blocks: usize, ) -> io::Result { for x in 0..tries { - transaction.sign(&[keypair], self.get_recent_blockhash()?); let mut buf = vec![0; serialized_size(&transaction).unwrap() as usize]; let mut wr = std::io::Cursor::new(&mut buf[..]); serialize_into(&mut wr, &transaction) @@ -96,47 +122,14 @@ impl ThinClient { return Ok(transaction.signatures[0]); } info!("{} tries failed transfer to {}", x, self.transactions_addr); + transaction.sign(keypairs, self.get_recent_blockhash()?); } Err(io::Error::new( io::ErrorKind::Other, - "retry_transfer failed", + format!("retry_transfer failed in {} retries", tries), )) } - /// Retry a sending a signed Transaction to the server for processing. - pub fn retry_transfer( - &self, - keypair: &Keypair, - transaction: &mut Transaction, - tries: usize, - ) -> io::Result { - for x in 0..tries { - transaction.sign(&[keypair], self.get_recent_blockhash()?); - let mut buf = vec![0; serialized_size(&transaction).unwrap() as usize]; - let mut wr = std::io::Cursor::new(&mut buf[..]); - serialize_into(&mut wr, &transaction) - .expect("serialize Transaction in pub fn transfer_signed"); - self.transactions_socket - .send_to(&buf[..], &self.transactions_addr)?; - if self.poll_for_signature(&transaction.signatures[0]).is_ok() { - return Ok(transaction.signatures[0]); - } - info!("{} tries failed transfer to {}", x, self.transactions_addr); - } - Err(io::Error::new( - io::ErrorKind::Other, - "retry_transfer failed", - )) - } - - pub fn get_account_data(&self, pubkey: &Pubkey) -> io::Result> { - self.rpc_client.get_account_data(pubkey) - } - - pub fn get_balance(&self, pubkey: &Pubkey) -> io::Result { - self.rpc_client.get_balance(pubkey) - } - pub fn get_transaction_count(&self) -> Result> { self.rpc_client.get_transaction_count() } @@ -198,6 +191,96 @@ impl ThinClient { } } +impl SyncClient for ThinClient { + fn send_message(&self, keypairs: &[&Keypair], message: Message) -> TransportResult { + let blockhash = self.get_recent_blockhash()?; + let mut transaction = Transaction::new(&keypairs, message, blockhash); + let signature = self.send_and_confirm_transaction(keypairs, &mut transaction, 5, 0)?; + Ok(signature) + } + + fn send_instruction( + &self, + keypair: &Keypair, + instruction: Instruction, + ) -> TransportResult { + let message = Message::new(vec![instruction]); + self.send_message(&[keypair], message) + } + + fn transfer( + &self, + lamports: u64, + keypair: &Keypair, + pubkey: &Pubkey, + ) -> TransportResult { + let transfer_instruction = + system_instruction::transfer(&keypair.pubkey(), pubkey, lamports); + self.send_instruction(keypair, transfer_instruction) + } + + fn get_account_data(&self, pubkey: &Pubkey) -> TransportResult>> { + Ok(self.rpc_client.get_account_data(pubkey).ok()) + } + + fn get_balance(&self, pubkey: &Pubkey) -> TransportResult { + let balance = self.rpc_client.get_balance(pubkey)?; + Ok(balance) + } + + fn get_signature_status( + &self, + signature: &Signature, + ) -> TransportResult>> { + let status = self + .rpc_client + .get_signature_status(&signature.to_string()) + .map_err(|err| { + io::Error::new( + io::ErrorKind::Other, + format!("send_transaction failed with error {}", err), + ) + })?; + Ok(status) + } +} + +impl AsyncClient for ThinClient { + fn async_send_transaction(&self, transaction: Transaction) -> io::Result { + let mut buf = vec![0; serialized_size(&transaction).unwrap() as usize]; + let mut wr = std::io::Cursor::new(&mut buf[..]); + serialize_into(&mut wr, &transaction) + .expect("serialize Transaction in pub fn transfer_signed"); + assert!(buf.len() < PACKET_DATA_SIZE); + self.transactions_socket + .send_to(&buf[..], &self.transactions_addr)?; + Ok(transaction.signatures[0]) + } + fn async_send_message(&self, keypairs: &[&Keypair], message: Message) -> io::Result { + let blockhash = self.get_recent_blockhash()?; + let transaction = Transaction::new(&keypairs, message, blockhash); + self.async_send_transaction(transaction) + } + fn async_send_instruction( + &self, + keypair: &Keypair, + instruction: Instruction, + ) -> io::Result { + let message = Message::new(vec![instruction]); + self.async_send_message(&[keypair], message) + } + fn async_transfer( + &self, + lamports: u64, + keypair: &Keypair, + pubkey: &Pubkey, + ) -> io::Result { + let transfer_instruction = + system_instruction::transfer(&keypair.pubkey(), pubkey, lamports); + self.async_send_instruction(keypair, transfer_instruction) + } +} + pub fn create_client((rpc, tpu): (SocketAddr, SocketAddr), range: (u16, u16)) -> ThinClient { let (_, transactions_socket) = solana_netutil::bind_in_range(range).unwrap(); ThinClient::new(rpc, tpu, transactions_socket) diff --git a/core/src/local_cluster.rs b/core/src/local_cluster.rs index 735553119..bfaf7d48c 100644 --- a/core/src/local_cluster.rs +++ b/core/src/local_cluster.rs @@ -11,6 +11,7 @@ use solana_client::thin_client::ThinClient; use solana_sdk::genesis_block::GenesisBlock; use solana_sdk::pubkey::Pubkey; use solana_sdk::signature::{Keypair, KeypairUtil}; +use solana_sdk::sync_client::SyncClient; use solana_sdk::system_transaction; use solana_sdk::timing::DEFAULT_SLOTS_PER_EPOCH; use solana_sdk::timing::DEFAULT_TICKS_PER_SLOT; @@ -363,7 +364,7 @@ impl LocalCluster { info!("Checking for vote account registration"); let vote_account_user_data = client.get_account_data(&vote_account_pubkey); - if let Ok(vote_account_user_data) = vote_account_user_data { + if let Ok(Some(vote_account_user_data)) = vote_account_user_data { if let Ok(vote_state) = VoteState::deserialize(&vote_account_user_data) { if vote_state.delegate_id == delegate_id { return Ok(()); diff --git a/core/src/storage_stage.rs b/core/src/storage_stage.rs index 104cb0fb9..cf4e96dea 100644 --- a/core/src/storage_stage.rs +++ b/core/src/storage_stage.rs @@ -16,6 +16,7 @@ use solana_client::thin_client::{create_client_with_timeout, ThinClient}; use solana_sdk::hash::Hash; use solana_sdk::pubkey::Pubkey; use solana_sdk::signature::{Keypair, KeypairUtil, Signature}; +use solana_sdk::sync_client::SyncClient; use solana_sdk::system_transaction; use solana_sdk::transaction::Transaction; use solana_storage_api::storage_instruction::{self, StorageInstruction};