blockworks-foundation
/
solana
mirror of https://github.com/blockworks-foundation/solana.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Remove record locks and parent locks from accounts (#4633) 

* Revert "Fix parent record locks usage in child banks (#4159)"

This reverts commit 69eeb7cf08.

* Revert "Fix DuplicateSignatures caused by races on frozen banks (#3819)"

This reverts commit 083090817a.

* Remove unused imports
This commit is contained in:
carllin 2019-06-10 22:05:46 -07:00 committed by GitHub
parent 807c69d97c
commit a4035a3c65
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 46 additions and 332 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
core/src/banking_stage.rs
View File

@ -369,26 +369,23 @@ impl BankingStage {
.collect()
}
fn record_transactions<'a, 'b>(
bank: &'a Bank,
txs: &'b [Transaction],
fn record_transactions(
bank_slot: u64,
txs: &[Transaction],
results: &[transaction::Result<()>],
poh: &Arc<Mutex<PohRecorder>>,
recordable_txs: &'b mut Vec<&'b Transaction>,
) -> Result<LockedAccountsResults<'a, 'b, &'b Transaction>> {
) -> Result<()> {
let processed_transactions: Vec<_> = results
.iter()
.zip(txs.iter())
.filter_map(|(r, x)| {
if Bank::can_commit(r) {
recordable_txs.push(x);
Some(x.clone())
} else {
None
}
})
.collect();
let record_locks = bank.lock_record_accounts(recordable_txs);
debug!("processed: {} ", processed_transactions.len());
// unlock all the accounts with errors which are filtered by the above `filter_map`
if !processed_transactions.is_empty() {
@ -400,16 +397,16 @@ impl BankingStage {
// record and unlock will unlock all the successful transactions
poh.lock()
.unwrap()
.record(bank.slot(), hash, processed_transactions)?;
.record(bank_slot, hash, processed_transactions)?;
}
Ok(record_locks)
Ok(())
}
fn process_and_record_transactions_locked(
bank: &Bank,
txs: &[Transaction],
poh: &Arc<Mutex<PohRecorder>>,
lock_results: &LockedAccountsResults<Transaction>,
lock_results: &LockedAccountsResults,
) -> Result<()> {
let now = Instant::now();
// Use a shorter maximum age when adding transactions into the pipeline. This will reduce
@ -422,12 +419,10 @@ impl BankingStage {
let freeze_lock = bank.freeze_lock();
let mut recordable_txs = vec![];
let (record_time, record_locks) = {
let record_time = {
let now = Instant::now();
let record_locks =
Self::record_transactions(bank, txs, &results, poh, &mut recordable_txs)?;
(now.elapsed(), record_locks)
Self::record_transactions(bank.slot(), txs, &results, poh)?;
now.elapsed()
};
let commit_time = {
@ -436,7 +431,6 @@ impl BankingStage {
now.elapsed()
};
drop(record_locks);
drop(freeze_lock);
debug!(
@ -1182,14 +1176,8 @@ mod tests {
];
let mut results = vec![Ok(()), Ok(())];
BankingStage::record_transactions(
&bank,
&transactions,
&results,
&poh_recorder,
&mut vec![],
)
.unwrap();
BankingStage::record_transactions(bank.slot(), &transactions, &results, &poh_recorder)
.unwrap();
let (_, entries) = entry_receiver.recv().unwrap();
assert_eq!(entries[0].0.transactions.len(), transactions.len());
@ -1198,27 +1186,15 @@ mod tests {
1,
InstructionError::new_result_with_negative_lamports(),
));
BankingStage::record_transactions(
&bank,
&transactions,
&results,
&poh_recorder,
&mut vec![],
)
.unwrap();
BankingStage::record_transactions(bank.slot(), &transactions, &results, &poh_recorder)
.unwrap();
let (_, entries) = entry_receiver.recv().unwrap();
assert_eq!(entries[0].0.transactions.len(), transactions.len());
// Other TransactionErrors should not be recorded
results[0] = Err(TransactionError::AccountNotFound);
BankingStage::record_transactions(
&bank,
&transactions,
&results,
&poh_recorder,
&mut vec![],
)
.unwrap();
BankingStage::record_transactions(bank.slot(), &transactions, &results, &poh_recorder)
.unwrap();
let (_, entries) = entry_receiver.recv().unwrap();
assert_eq!(entries[0].0.transactions.len(), transactions.len() - 1);
}

6
core/src/blocktree_processor.rs
View File

@ -11,7 +11,6 @@ use solana_sdk::genesis_block::GenesisBlock;
use solana_sdk::timing::duration_as_ms;
use solana_sdk::timing::MAX_RECENT_BLOCKHASHES;
use solana_sdk::transaction::Result;
use solana_sdk::transaction::Transaction;
use std::result;
use std::sync::Arc;
use std::time::{Duration, Instant};
@ -33,10 +32,7 @@ fn first_err(results: &[Result<()>]) -> Result<()> {
Ok(())
}
fn par_execute_entries(
bank: &Bank,
entries: &[(&Entry, LockedAccountsResults<Transaction>)],
) -> Result<()> {
fn par_execute_entries(bank: &Bank, entries: &[(&Entry, LockedAccountsResults)]) -> Result<()> {
inc_new_counter_debug!("bank-par_execute_entries-count", entries.len());
let results: Vec<Result<()>> = PAR_THREAD_POOL.with(|thread_pool| {
thread_pool.borrow().install(|| {

227
runtime/src/accounts.rs
View File

@ -19,38 +19,16 @@ use solana_sdk::syscall;
use solana_sdk::system_program;
use solana_sdk::transaction::Result;
use solana_sdk::transaction::{Transaction, TransactionError};
use std::borrow::Borrow;
use std::collections::{HashMap, HashSet};
use std::env;
use std::fs::remove_dir_all;
use std::iter::once;
use std::ops::Neg;
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::sleep;
use std::time::Duration;
const ACCOUNTSDB_DIR: &str = "accountsdb";
const NUM_ACCOUNT_DIRS: usize = 4;
const WAIT_FOR_PARENT_MS: u64 = 5;
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum AccountLockType {
AccountLock,
RecordLock,
}
type AccountLocks = Mutex<HashSet<Pubkey>>;
// Locks for accounts that are currently being recorded + committed
type RecordLocks = (
// Record Locks for the current bank
Arc<AccountLocks>,
// Any unreleased record locks from all parent/grandparent banks. We use Arc<Mutex> to
// avoid copies when calling new_from_parent().
Vec<Arc<AccountLocks>>,
);
/// This structure handles synchronization for db
#[derive(Default, Debug, Serialize, Deserialize)]
@ -60,12 +38,7 @@ pub struct Accounts {
pub accounts_db: Arc<AccountsDB>,
/// set of accounts which are currently in the pipeline
#[serde(skip)]
account_locks: AccountLocks,
/// set of accounts which are about to record + commit
#[serde(skip)]
record_locks: Mutex<RecordLocks>,
account_locks: Mutex<HashSet<Pubkey>>,
/// List of persistent stores
paths: String,
@ -126,38 +99,15 @@ impl Accounts {
Accounts {
accounts_db,
account_locks: Mutex::new(HashSet::new()),
record_locks: Mutex::new((Arc::new(Mutex::new(HashSet::new())), vec![])),
paths,
own_paths,
}
}
pub fn new_from_parent(parent: &Accounts) -> Self {
let accounts_db = parent.accounts_db.clone();
let parent_record_locks: Vec<_> = {
let (ref record_locks, ref mut grandparent_record_locks) =
*parent.record_locks.lock().unwrap();
// Note that when creating a child bank, we only care about the locks that are held for
// accounts that are in txs that are currently recording + committing, because other
// incoming txs on this bank that are not yet recording will not make it to bank commit.
//
// Thus:
// 1) The child doesn't need to care about potential "future" account locks on its parent
// bank that the parent does not currently hold.
// 2) The child only needs the currently held "record locks" from the parent.
// 3) The parent doesn't need to retain any of the locks other than the ones it owns so
// that unlock() can be called later (the grandparent locks can be given to the child).
once(record_locks.clone())
.chain(grandparent_record_locks.drain(..))
.filter(|a| !a.lock().unwrap().is_empty())
.collect()
};
Accounts {
accounts_db,
account_locks: Mutex::new(HashSet::new()),
record_locks: Mutex::new((Arc::new(Mutex::new(HashSet::new())), parent_record_locks)),
paths: parent.paths.clone(),
own_paths: parent.own_paths,
}
@ -365,63 +315,24 @@ impl Accounts {
}
fn lock_account(
(fork_locks, parent_locks): (&mut HashSet<Pubkey>, &mut Vec<Arc<AccountLocks>>),
locks: &mut HashSet<Pubkey>,
keys: &[&Pubkey],
error_counters: &mut ErrorCounters,
) -> Result<()> {
// Copy all the accounts
for k in keys {
let is_locked = {
if fork_locks.contains(k) {
true
} else {
// Check parent locks. As soon as a set of parent locks is empty,
// we can remove it from the list b/c that means the parent has
// released the locks.
parent_locks.retain(|p| {
loop {
{
// If a parent bank holds a record lock for this account, then loop
// until that lock is released
let p = p.lock().unwrap();
if !p.contains(k) {
break;
}
}
// If a parent is currently recording for this key, then drop lock and wait
sleep(Duration::from_millis(WAIT_FOR_PARENT_MS));
}
let p = p.lock().unwrap();
!p.is_empty()
});
false
}
};
if is_locked {
if locks.contains(k) {
error_counters.account_in_use += 1;
debug!("Account in use: {:?}", k);
return Err(TransactionError::AccountInUse);
}
}
for k in keys {
fork_locks.insert(**k);
locks.insert(**k);
}
Ok(())
}
fn lock_record_account(record_locks: &AccountLocks, keys: &[&Pubkey]) {
let mut fork_locks = record_locks.lock().unwrap();
for k in keys {
// The fork locks should always be a subset of the account locks, so
// the account locks should prevent record locks from ever touching the
// same accounts
assert!(!fork_locks.contains(k));
fork_locks.insert(**k);
}
}
fn unlock_account(tx: &Transaction, result: &Result<()>, locks: &mut HashSet<Pubkey>) {
match result {
Err(TransactionError::AccountInUse) => (),
@ -433,15 +344,6 @@ impl Accounts {
}
}
fn unlock_record_account<I>(tx: &I, locks: &mut HashSet<Pubkey>)
where
I: Borrow<Transaction>,
{
for k in &tx.borrow().message().account_keys {
locks.remove(k);
}
}
fn hash_account(stored_account: &StoredAccount) -> Hash {
let mut hasher = Hasher::default();
hasher.hash(&serialize(&stored_account.balance).unwrap());
@ -481,18 +383,14 @@ impl Accounts {
/// This function will prevent multiple threads from modifying the same account state at the
/// same time
#[must_use]
pub fn lock_accounts<I>(&self, txs: &[I]) -> Vec<Result<()>>
where
I: Borrow<Transaction>,
{
let (_, ref mut parent_record_locks) = *self.record_locks.lock().unwrap();
pub fn lock_accounts(&self, txs: &[Transaction]) -> Vec<Result<()>> {
let mut error_counters = ErrorCounters::default();
let rv = txs
.iter()
.map(|tx| {
let message = tx.borrow().message();
let message = &tx.message();
Self::lock_account(
(&mut self.account_locks.lock().unwrap(), parent_record_locks),
&mut self.account_locks.lock().unwrap(),
&message.get_account_keys_by_lock_type().0,
&mut error_counters,
)
@ -509,37 +407,13 @@ impl Accounts {
rv
}
pub fn lock_record_accounts<I>(&self, txs: &[I])
where
I: Borrow<Transaction>,
{
let record_locks = self.record_locks.lock().unwrap();
for tx in txs {
let message = tx.borrow().message();
Self::lock_record_account(&record_locks.0, &message.get_account_keys_by_lock_type().0);
}
}
/// Once accounts are unlocked, new transactions that modify that state can enter the pipeline
pub fn unlock_accounts<I>(&self, txs: &[I], results: &[Result<()>])
where
I: Borrow<Transaction>,
{
let my_locks = &mut self.account_locks.lock().unwrap();
pub fn unlock_accounts(&self, txs: &[Transaction], results: &[Result<()>]) {
let mut account_locks = self.account_locks.lock().unwrap();
debug!("bank unlock accounts");
txs.iter()
.zip(results.iter())
.for_each(|(tx, result)| Self::unlock_account(tx.borrow(), result, my_locks));
}
pub fn unlock_record_accounts<I>(&self, txs: &[I])
where
I: Borrow<Transaction>,
{
let (ref my_record_locks, _) = *self.record_locks.lock().unwrap();
for tx in txs {
Self::unlock_record_account(tx, &mut my_record_locks.lock().unwrap())
}
.for_each(|(tx, result)| Self::unlock_account(tx, result, &mut account_locks));
}
pub fn has_accounts(&self, fork: Fork) -> bool {
@ -616,8 +490,6 @@ mod tests {
use solana_sdk::syscall;
use solana_sdk::transaction::Transaction;
use std::io::Cursor;
use std::thread::{sleep, Builder};
use std::time::Duration;
fn load_accounts_with_fee(
tx: Transaction,
@ -1109,85 +981,6 @@ mod tests {
assert_eq!(accounts.hash_internal_state(0), None);
}
#[test]
fn test_parent_locked_record_accounts() {
let mut parent = Accounts::new(None);
let locked_pubkey = Keypair::new().pubkey();
let mut locked_accounts = HashSet::new();
locked_accounts.insert(locked_pubkey);
parent.record_locks = Mutex::new((Arc::new(Mutex::new(locked_accounts.clone())), vec![]));
let parent = Arc::new(parent);
let child = Accounts::new_from_parent(&parent);
// Make sure child record locks contains the parent's locked record accounts
{
let (_, ref parent_record_locks) = *child.record_locks.lock().unwrap();
assert_eq!(parent_record_locks.len(), 1);
assert_eq!(locked_accounts, *parent_record_locks[0].lock().unwrap());
}
let parent_ = parent.clone();
let parent_thread = Builder::new()
.name("exit".to_string())
.spawn(move || {
sleep(Duration::from_secs(2));
// Unlock the accounts in the parent
{
let (ref parent_record_locks, _) = *parent_.record_locks.lock().unwrap();
parent_record_locks.lock().unwrap().clear();
}
})
.unwrap();
// Function will block until the parent_thread unlocks the parent's record lock
assert_eq!(
Accounts::lock_account(
(
&mut child.account_locks.lock().unwrap(),
&mut child.record_locks.lock().unwrap().1
),
&vec![&locked_pubkey],
&mut ErrorCounters::default()
),
Ok(())
);
// Make sure that the function blocked
parent_thread.join().unwrap();
{
// Check the lock was successfully obtained
let child_account_locks = &mut child.account_locks.lock().unwrap();
let parent_record_locks = child.record_locks.lock().unwrap();
assert_eq!(child_account_locks.len(), 1);
// Make sure child removed the parent's record locks after the parent had
// released all its locks
assert!(parent_record_locks.1.is_empty());
// After all the checks pass, clear the account we just locked from the
// set of locks
child_account_locks.clear();
}
// Make sure calling new_from_parent() on the child bank also cleans up the copy of old locked
// parent accounts, in case the child doesn't call lock_account() after a parent has
// released their account locks
{
// Mock an empty set of parent record accounts in the child bank
let (_, ref mut parent_record_locks) = *child.record_locks.lock().unwrap();
parent_record_locks.push(Arc::new(Mutex::new(HashSet::new())));
}
// Call new_from_parent, make sure the empty parent locked_accounts is purged
let child2 = Accounts::new_from_parent(&child);
{
let (_, ref mut parent_record_locks) = *child.record_locks.lock().unwrap();
assert!(parent_record_locks.is_empty());
let (_, ref mut parent_record_locks2) = *child2.record_locks.lock().unwrap();
assert!(parent_record_locks2.is_empty());
}
}
fn create_accounts(accounts: &Accounts, pubkeys: &mut Vec<Pubkey>, num: usize) {
for t in 0..num {
let pubkey = Pubkey::new_rand();

51
runtime/src/bank.rs
View File

@ -2,7 +2,7 @@
//! programs. It offers a high-level API that signs transactions
//! on behalf of the caller, and a low-level API for when they have
//! already been signed and verified.
use crate::accounts::{AccountLockType, Accounts};
use crate::accounts::Accounts;
use crate::accounts_db::{
AccountsDB, ErrorCounters, InstructionAccounts, InstructionCredits, InstructionLoaders,
};
@ -38,7 +38,6 @@ use solana_sdk::syscall::tick_height::{self, TickHeight};
use solana_sdk::system_transaction;
use solana_sdk::timing::{duration_as_ms, duration_as_us, MAX_RECENT_BLOCKHASHES};
use solana_sdk::transaction::{Result, Transaction, TransactionError};
use std::borrow::Borrow;
use std::cmp;
use std::collections::HashMap;
use std::fmt;
@ -629,53 +628,29 @@ impl Bank {
.map_or(Ok(()), |sig| self.get_signature_status(sig).unwrap())
}
pub fn lock_accounts<'a, 'b, I>(&'a self, txs: &'b [I]) -> LockedAccountsResults<'a, 'b, I>
where
I: std::borrow::Borrow<Transaction>,
{
pub fn lock_accounts<'a, 'b>(
&'a self,
txs: &'b [Transaction],
) -> LockedAccountsResults<'a, 'b> {
if self.is_frozen() {
warn!("=========== FIXME: lock_accounts() working on a frozen bank! ================");
}
// TODO: put this assert back in
// assert!(!self.is_frozen());
let results = self.rc.accounts.lock_accounts(txs);
LockedAccountsResults::new(results, &self, txs, AccountLockType::AccountLock)
LockedAccountsResults::new(results, &self, txs)
}
pub fn unlock_accounts<I>(&self, locked_accounts_results: &mut LockedAccountsResults<I>)
where
I: Borrow<Transaction>,
{
pub fn unlock_accounts(&self, locked_accounts_results: &mut LockedAccountsResults) {
if locked_accounts_results.needs_unlock {
locked_accounts_results.needs_unlock = false;
match locked_accounts_results.lock_type() {
AccountLockType::AccountLock => self.rc.accounts.unlock_accounts(
locked_accounts_results.transactions(),
locked_accounts_results.locked_accounts_results(),
),
AccountLockType::RecordLock => self
.rc
.accounts
.unlock_record_accounts(locked_accounts_results.transactions()),
}
self.rc.accounts.unlock_accounts(
locked_accounts_results.transactions(),
locked_accounts_results.locked_accounts_results(),
)
}
}
pub fn lock_record_accounts<'a, 'b, I>(
&'a self,
txs: &'b [I],
) -> LockedAccountsResults<'a, 'b, I>
where
I: std::borrow::Borrow<Transaction>,
{
self.rc.accounts.lock_record_accounts(txs);
LockedAccountsResults::new(vec![], &self, txs, AccountLockType::RecordLock)
}
pub fn unlock_record_accounts(&self, txs: &[Transaction]) {
self.rc.accounts.unlock_record_accounts(txs)
}
fn load_accounts(
&self,
txs: &[Transaction],
@ -836,7 +811,7 @@ impl Bank {
pub fn load_and_execute_transactions(
&self,
txs: &[Transaction],
lock_results: &LockedAccountsResults<Transaction>,
lock_results: &LockedAccountsResults,
max_age: usize,
) -> (
Vec<Result<(InstructionAccounts, InstructionLoaders, InstructionCredits)>>,
@ -978,7 +953,7 @@ impl Bank {
pub fn load_execute_and_commit_transactions(
&self,
txs: &[Transaction],
lock_results: &LockedAccountsResults<Transaction>,
lock_results: &LockedAccountsResults,
max_age: usize,
) -> Vec<Result<()>> {
let (loaded_accounts, executed) =

38
runtime/src/locked_accounts_results.rs
View File

@ -1,48 +1,39 @@
use crate::accounts::AccountLockType;
use crate::bank::Bank;
use solana_sdk::transaction::{Result, Transaction};
use std::borrow::Borrow;
// Represents the results of trying to lock a set of accounts
pub struct LockedAccountsResults<'a, 'b, I: Borrow<Transaction>> {
pub struct LockedAccountsResults<'a, 'b> {
locked_accounts_results: Vec<Result<()>>,
bank: &'a Bank,
transactions: &'b [I],
lock_type: AccountLockType,
transactions: &'b [Transaction],
pub(crate) needs_unlock: bool,
}
impl<'a, 'b, I: Borrow<Transaction>> LockedAccountsResults<'a, 'b, I> {
impl<'a, 'b> LockedAccountsResults<'a, 'b> {
pub fn new(
locked_accounts_results: Vec<Result<()>>,
bank: &'a Bank,
transactions: &'b [I],
lock_type: AccountLockType,
transactions: &'b [Transaction],
) -> Self {
Self {
locked_accounts_results,
bank,
transactions,
needs_unlock: true,
lock_type,
}
}
pub fn lock_type(&self) -> AccountLockType {
self.lock_type.clone()
}
pub fn locked_accounts_results(&self) -> &Vec<Result<()>> {
&self.locked_accounts_results
}
pub fn transactions(&self) -> &[I] {
pub fn transactions(&self) -> &[Transaction] {
self.transactions
}
}
// Unlock all locked accounts in destructor.
impl<'a, 'b, I: Borrow<Transaction>> Drop for LockedAccountsResults<'a, 'b, I> {
impl<'a, 'b> Drop for LockedAccountsResults<'a, 'b> {
fn drop(&mut self) {
if self.needs_unlock {
self.bank.unlock_accounts(self)
@ -89,23 +80,6 @@ mod tests {
.all(|x| x.is_ok()));
}
#[test]
fn test_record_locks() {
let (bank, txs) = setup();
// Test getting record locks
let lock_results = bank.lock_record_accounts(&txs);
// Grabbing record locks doesn't return any results, must succeed or panic.
assert!(lock_results.locked_accounts_results().is_empty());
drop(lock_results);
// Now grabbing record locks should work again
let lock_results2 = bank.lock_record_accounts(&txs);
assert!(lock_results2.locked_accounts_results().is_empty());
}
fn setup() -> (Bank, Vec<Transaction>) {
let dummy_leader_pubkey = Pubkey::new_rand();
let GenesisBlockInfo {