gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

transaction batch (#5962)

Browse Source 
* transaction batch

* fixup
This commit is contained in:
Rob Walker
2019-09-19 10:06:08 -07:00
committed by GitHub
parent 1a71804ef2
commit 5cbd1190b2
5 changed files with 184 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
core/src/banking_stage.rs
View File

@ -1,42 +1,45 @@
//! The `banking_stage` processes Transaction messages. It is intended to be used
//! to contruct a software pipeline. The stage uses all available CPU cores and
//! can do its processing in parallel with signature verification on the GPU.
use crate::blocktree::Blocktree;
use crate::cluster_info::ClusterInfo;
use crate::entry::hash_transactions;
use crate::leader_schedule_cache::LeaderScheduleCache;
use crate::packet::PACKETS_PER_BATCH;
use crate::packet::{Packet, Packets};
use crate::poh_recorder::{PohRecorder, PohRecorderError, WorkingBankEntry};
use crate::poh_service::PohService;
use crate::result::{Error, Result};
use crate::service::Service;
use crate::sigverify_stage::VerifiedPackets;
use crate::{
blocktree::Blocktree,
cluster_info::ClusterInfo,
entry::hash_transactions,
leader_schedule_cache::LeaderScheduleCache,
packet::PACKETS_PER_BATCH,
packet::{Packet, Packets},
poh_recorder::{PohRecorder, PohRecorderError, WorkingBankEntry},
poh_service::PohService,
result::{Error, Result},
service::Service,
sigverify_stage::VerifiedPackets,
};
use bincode::deserialize;
use crossbeam_channel::{Receiver as CrossbeamReceiver, RecvTimeoutError};
use itertools::Itertools;
use solana_measure::measure::Measure;
use solana_metrics::{inc_new_counter_debug, inc_new_counter_info, inc_new_counter_warn};
use solana_runtime::accounts_db::ErrorCounters;
use solana_runtime::bank::Bank;
use solana_runtime::locked_accounts_results::LockedAccountsResults;
use solana_sdk::clock::{
DEFAULT_TICKS_PER_SECOND, DEFAULT_TICKS_PER_SLOT, MAX_PROCESSING_AGE,
MAX_TRANSACTION_FORWARDING_DELAY,
use solana_runtime::{accounts_db::ErrorCounters, bank::Bank, transaction_batch::TransactionBatch};
use solana_sdk::{
clock::{
DEFAULT_TICKS_PER_SECOND, DEFAULT_TICKS_PER_SLOT, MAX_PROCESSING_AGE,
MAX_TRANSACTION_FORWARDING_DELAY,
},
poh_config::PohConfig,
pubkey::Pubkey,
timing::{duration_as_ms, timestamp},
transaction::{self, Transaction, TransactionError},
};
use std::{
cmp, env,
net::UdpSocket,
sync::atomic::AtomicBool,
sync::mpsc::Receiver,
sync::{Arc, Mutex, RwLock},
thread::{self, Builder, JoinHandle},
time::Duration,
time::Instant,
};
use solana_sdk::poh_config::PohConfig;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::timing::{duration_as_ms, timestamp};
use solana_sdk::transaction::{self, Transaction, TransactionError};
use std::cmp;
use std::env;
use std::net::UdpSocket;
use std::sync::atomic::AtomicBool;
use std::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex, RwLock};
use std::thread::{self, Builder, JoinHandle};
use std::time::Duration;
use std::time::Instant;
type PacketsAndOffsets = (Packets, Vec<usize>);
pub type UnprocessedPackets = Vec<PacketsAndOffsets>;
@ -481,16 +484,16 @@ impl BankingStage {
fn process_and_record_transactions_locked(
bank: &Bank,
poh: &Arc<Mutex<PohRecorder>>,
lock_results: &LockedAccountsResults,
batch: &TransactionBatch,
) -> (Result<usize>, Vec<usize>) {
let mut load_execute_time = Measure::start("load_execute_time");
// Use a shorter maximum age when adding transactions into the pipeline. This will reduce
// the likelihood of any single thread getting starved and processing old ids.
// TODO: Banking stage threads should be prioritized to complete faster then this queue
// expires.
let txs = lock_results.transactions();
let txs = batch.transactions();
let (mut loaded_accounts, results, mut retryable_txs, tx_count, signature_count) =
bank.load_and_execute_transactions(lock_results, MAX_PROCESSING_AGE);
bank.load_and_execute_transactions(batch, MAX_PROCESSING_AGE);
load_execute_time.stop();
let freeze_lock = bank.freeze_lock();
@ -543,16 +546,16 @@ impl BankingStage {
let mut lock_time = Measure::start("lock_time");
// Once accounts are locked, other threads cannot encode transactions that will modify the
// same account state
let lock_results = bank.lock_accounts(txs, None);
let batch = bank.prepare_batch(txs, None);
lock_time.stop();
let (result, mut retryable_txs) =
Self::process_and_record_transactions_locked(bank, poh, &lock_results);
Self::process_and_record_transactions_locked(bank, poh, &batch);
retryable_txs.iter_mut().for_each(|x| *x += chunk_offset);
let mut unlock_time = Measure::start("unlock_time");
// Once the accounts are new transactions can enter the pipeline to process them
drop(lock_results);
drop(batch);
unlock_time.stop();
debug!(

104
core/src/blocktree_processor.rs
View File

@ -8,7 +8,7 @@ use rayon::prelude::*;
use rayon::ThreadPool;
use solana_metrics::{datapoint, datapoint_error, inc_new_counter_debug};
use solana_runtime::bank::Bank;
use solana_runtime::locked_accounts_results::LockedAccountsResults;
use solana_runtime::transaction_batch::TransactionBatch;
use solana_sdk::clock::{Slot, MAX_RECENT_BLOCKHASHES};
use solana_sdk::genesis_block::GenesisBlock;
use solana_sdk::hash::Hash;
@ -37,34 +37,41 @@ fn first_err(results: &[Result<()>]) -> Result<()> {
Ok(())
}
fn par_execute_entries(bank: &Bank, entries: &[(&Entry, LockedAccountsResults)]) -> Result<()> {
inc_new_counter_debug!("bank-par_execute_entries-count", entries.len());
fn execute_batch(batch: &TransactionBatch) -> Result<()> {
let results = batch
.bank()
.load_execute_and_commit_transactions(batch, MAX_RECENT_BLOCKHASHES);
let mut first_err = None;
for (result, transaction) in results.iter().zip(batch.transactions()) {
if let Err(ref err) = result {
if first_err.is_none() {
first_err = Some(result.clone());
}
warn!(
"Unexpected validator error: {:?}, transaction: {:?}",
err, transaction
);
datapoint_error!(
"validator_process_entry_error",
(
"error",
format!("error: {:?}, transaction: {:?}", err, transaction),
String
)
);
}
}
first_err.unwrap_or(Ok(()))
}
fn execute_batches(batches: &[TransactionBatch]) -> Result<()> {
inc_new_counter_debug!("bank-par_execute_entries-count", batches.len());
let results: Vec<Result<()>> = PAR_THREAD_POOL.with(|thread_pool| {
thread_pool.borrow().install(|| {
entries
batches
.into_par_iter()
.map(|(entry, locked_accounts)| {
let results = bank.load_execute_and_commit_transactions(
locked_accounts,
MAX_RECENT_BLOCKHASHES,
);
let mut first_err = None;
for (r, tx) in results.iter().zip(entry.transactions.iter()) {
if let Err(ref entry) = r {
if first_err.is_none() {
first_err = Some(r.clone());
}
if !Bank::can_commit(&r) {
warn!("Unexpected validator error: {:?}, tx: {:?}", entry, tx);
datapoint_error!(
"validator_process_entry_error",
("error", format!("error: {:?}, tx: {:?}", entry, tx), String)
);
}
}
}
first_err.unwrap_or(Ok(()))
})
.map(|batch| execute_batch(batch))
.collect()
})
});
@ -77,45 +84,40 @@ fn par_execute_entries(bank: &Bank, entries: &[(&Entry, LockedAccountsResults)])
/// 2. Process the locked group in parallel
/// 3. Register the `Tick` if it's available
/// 4. Update the leader scheduler, goto 1
pub fn process_entries(
bank: &Bank,
entries: &[Entry],
randomize_tx_execution_order: bool,
) -> Result<()> {
pub fn process_entries(bank: &Bank, entries: &[Entry], randomize: bool) -> Result<()> {
// accumulator for entries that can be processed in parallel
let mut mt_group = vec![];
let mut batches = vec![];
for entry in entries {
if entry.is_tick() {
// if its a tick, execute the group and register the tick
par_execute_entries(bank, &mt_group)?;
mt_group = vec![];
execute_batches(&batches)?;
batches.clear();
bank.register_tick(&entry.hash);
continue;
}
// else loop on processing the entry
loop {
let txs_execution_order = if randomize_tx_execution_order {
let mut random_txs_execution_order: Vec<usize> =
(0..entry.transactions.len()).collect();
random_txs_execution_order.shuffle(&mut thread_rng());
Some(random_txs_execution_order)
let iteration_order = if randomize {
let mut iteration_order: Vec<usize> = (0..entry.transactions.len()).collect();
iteration_order.shuffle(&mut thread_rng());
Some(iteration_order)
} else {
None
};
// try to lock the accounts
let lock_results = bank.lock_accounts(&entry.transactions, txs_execution_order);
let batch = bank.prepare_batch(&entry.transactions, iteration_order);
let first_lock_err = first_err(lock_results.locked_accounts_results());
let first_lock_err = first_err(batch.lock_results());
// if locking worked
if first_lock_err.is_ok() {
mt_group.push((entry, lock_results));
batches.push(batch);
// done with this entry
break;
}
// else we failed to lock, 2 possible reasons
if mt_group.is_empty() {
if batches.is_empty() {
// An entry has account lock conflicts with *itself*, which should not happen
// if generated by a properly functioning leader
datapoint!(
@ -134,12 +136,12 @@ pub fn process_entries(
} else {
// else we have an entry that conflicts with a prior entry
// execute the current queue and try to process this entry again
par_execute_entries(bank, &mt_group)?;
mt_group = vec![];
execute_batches(&batches)?;
batches.clear();
}
}
}
par_execute_entries(bank, &mt_group)?;
execute_batches(&batches)?;
Ok(())
}
@ -1070,14 +1072,14 @@ pub mod tests {
// Check all accounts are unlocked
let txs1 = &entry_1_to_mint.transactions[..];
let txs2 = &entry_2_to_3_mint_to_1.transactions[..];
let locked_accounts1 = bank.lock_accounts(txs1, None);
for result in locked_accounts1.locked_accounts_results() {
let batch1 = bank.prepare_batch(txs1, None);
for result in batch1.lock_results() {
assert!(result.is_ok());
}
// txs1 and txs2 have accounts that conflict, so we must drop txs1 first
drop(locked_accounts1);
let locked_accounts2 = bank.lock_accounts(txs2, None);
for result in locked_accounts2.locked_accounts_results() {
drop(batch1);
let batch2 = bank.prepare_batch(txs2, None);
for result in batch2.lock_results() {
assert!(result.is_ok());
}
}