Skip to content

Avoid querying unwritten values from changesets during historical lookups #2966

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gakonst merged 1 commit into from
Jun 5, 2023
Merged

Avoid querying unwritten values from changesets during historical lookups #2966

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
202 changes: 135 additions & 67 deletions crates/storage/provider/src/providers/state/historical.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,46 +31,93 @@ pub struct HistoricalStateProviderRef<'a, 'b, TX: DbTx<'a>> {
_phantom: PhantomData<&'a TX>,
}

pub enum HistoryInfo {
NotWritten,
InChangeset(u64),
InPlainState,
}

impl<'a, 'b, TX: DbTx<'a>> HistoricalStateProviderRef<'a, 'b, TX> {
/// Create new StateProvider from history transaction number
pub fn new(tx: &'b TX, block_number: BlockNumber) -> Self {
Self { tx, block_number, _phantom: PhantomData {} }
}
}
impl<'a, 'b, TX: DbTx<'a>> AccountProvider for HistoricalStateProviderRef<'a, 'b, TX> {
/// Get basic account information.
fn basic_account(&self, address: Address) -> Result<Option<Account>> {

/// Lookup an account in the AccountHistory table
pub fn account_history_lookup(&self, address: Address) -> Result<HistoryInfo> {
// history key to search IntegerList of block number changesets.
let history_key = ShardedKey::new(address, self.block_number);
let mut cursor = self.tx.cursor_read::<tables::AccountHistory>()?;

if let Some(chunk) =
cursor.seek(history_key)?.filter(|(key, _)| key.key == address).map(|x| x.1 .0)
{
let chunk = chunk.enable_rank();
let rank = chunk.rank(self.block_number as usize);
if rank == 0 && !cursor.prev()?.is_some_and(|(key, _)| key.key == address) {
return Ok(HistoryInfo::NotWritten)
}
if rank < chunk.len() {
Ok(HistoryInfo::InChangeset(chunk.select(rank) as u64))
} else {
Ok(HistoryInfo::InPlainState)
}
} else {
Ok(HistoryInfo::NotWritten)
}
Comment on lines +52 to +67
Copy link
Copy Markdown
Member

@gakonst gakonst Jun 5, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It could be helpful if we had some inline ascii diagram showing how this works, I think basically it says:

If there's a value in the history table then it is:

  • Not Written: If there are 0 elements in that entry AND there exists a previous entry in the table for that key, meaning it was previously written and now there was no change. We use this later for the optimization in the changeset read. In practice this is going to be rare as in most cases it will berank != 0 and the if statement will short circuit.
  • InChangeset: If the number of items less than block number in that bucket is less than the items in the bucket, return the corresponding changeset key.
  • InPlainState: If it's larger than the bucket len, then it means it's in the plain state

If there's no value, also return NotWritten.

}

/// Lookup a storage key in the StorageHistory table
pub fn storage_history_lookup(
&self,
address: Address,
storage_key: StorageKey,
) -> Result<HistoryInfo> {
// history key to search IntegerList of block number changesets.
let history_key = StorageShardedKey::new(address, storage_key, self.block_number);
let mut cursor = self.tx.cursor_read::<tables::StorageHistory>()?;
Copy link
Copy Markdown
Member

@gakonst gakonst Jun 5, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same idea as above here but with the duptable making it a little more verbose


let changeset_block_number = self
.tx
.cursor_read::<tables::AccountHistory>()?
if let Some(chunk) = cursor
.seek(history_key)?
.filter(|(key, _)| key.key == address)
.map(|(_, list)| {
list.0.enable_rank().successor(self.block_number as usize).map(|i| i as u64)
});

// if changeset of the block is present we are getting value from that changeset
if let Some(Some(changeset_block_number)) = changeset_block_number {
let account = self
.filter(|(key, _)| key.address == address && key.sharded_key.key == storage_key)
.map(|x| x.1 .0)
{
let chunk = chunk.enable_rank();
let rank = chunk.rank(self.block_number as usize);
if rank == 0 &&
!cursor.prev()?.is_some_and(|(key, _)| {
key.address == address && key.sharded_key.key == storage_key
})
{
return Ok(HistoryInfo::NotWritten)
}
if rank < chunk.len() {
Ok(HistoryInfo::InChangeset(chunk.select(rank) as u64))
} else {
Ok(HistoryInfo::InPlainState)
}
} else {
Ok(HistoryInfo::NotWritten)
}
}
}

impl<'a, 'b, TX: DbTx<'a>> AccountProvider for HistoricalStateProviderRef<'a, 'b, TX> {
/// Get basic account information.
fn basic_account(&self, address: Address) -> Result<Option<Account>> {
match self.account_history_lookup(address)? {
HistoryInfo::NotWritten => Ok(None),
HistoryInfo::InChangeset(changeset_block_number) => Ok(self
.tx
.cursor_dup_read::<tables::AccountChangeSet>()?
.seek_by_key_subkey(changeset_block_number, address)?
.filter(|acc| acc.address == address)
.ok_or(ProviderError::AccountChangesetNotFound {
block_number: changeset_block_number,
address,
})?;
Ok(account.info)
} else if changeset_block_number.is_none() {
// if there is no shard, return empty account.
Ok(None)
} else {
// if changeset is not present that means that there was history shard but we need to
// use newest value from plain state. Or zero if none.
Ok(self.tx.get::<tables::PlainAccountState>(address)?)
})?
.info),
HistoryInfo::InPlainState => Ok(self.tx.get::<tables::PlainAccountState>(address)?),
}
}
}
Expand Down Expand Up @@ -104,43 +151,27 @@ impl<'a, 'b, TX: DbTx<'a>> StateRootProvider for HistoricalStateProviderRef<'a,
impl<'a, 'b, TX: DbTx<'a>> StateProvider for HistoricalStateProviderRef<'a, 'b, TX> {
/// Get storage.
fn storage(&self, address: Address, storage_key: StorageKey) -> Result<Option<StorageValue>> {
// history key to search IntegerList of block changesets.
let history_key = StorageShardedKey::new(address, storage_key, self.block_number);

let changeset_block_number = self
.tx
.cursor_read::<tables::StorageHistory>()?
.seek(history_key)?
.filter(|(key, _)| key.address == address && key.sharded_key.key == storage_key)
.map(|(_, list)| {
list.0.enable_rank().successor(self.block_number as usize).map(|i| i as u64)
});

// if changeset transition id is present we are getting value from changeset
if let Some(Some(changeset_block_number)) = changeset_block_number {
let storage_entry = self
.tx
.cursor_dup_read::<tables::StorageChangeSet>()?
.seek_by_key_subkey((changeset_block_number, address).into(), storage_key)?
.filter(|entry| entry.key == storage_key)
.ok_or(ProviderError::StorageChangesetNotFound {
block_number: changeset_block_number,
address,
storage_key,
})?;
Ok(Some(storage_entry.value))
} else if changeset_block_number.is_none() {
// if there is no shards, return empty account.
return Ok(None)
} else {
// if changeset is not present that means that there was history shard but we need to
// use newest value from plain state
Ok(self
match self.storage_history_lookup(address, storage_key)? {
HistoryInfo::NotWritten => Ok(None),
HistoryInfo::InChangeset(changeset_block_number) => Ok(Some(
self.tx
.cursor_dup_read::<tables::StorageChangeSet>()?
.seek_by_key_subkey((changeset_block_number, address).into(), storage_key)?
.filter(|entry| entry.key == storage_key)
.ok_or(ProviderError::StorageChangesetNotFound {
block_number: changeset_block_number,
address,
storage_key,
})?
.value,
)),
HistoryInfo::InPlainState => Ok(self
.tx
.cursor_dup_read::<tables::PlainStorageState>()?
.seek_by_key_subkey(address, storage_key)?
.filter(|entry| entry.key == storage_key)
.map(|entry| entry.value))
.map(|entry| entry.value)
.or(Some(StorageValue::ZERO))),
}
}

Expand Down Expand Up @@ -201,6 +232,7 @@ mod tests {
use reth_primitives::{hex_literal::hex, Account, StorageEntry, H160, H256, U256};

const ADDRESS: H160 = H160(hex!("0000000000000000000000000000000000000001"));
const HIGHER_ADDRESS: H160 = H160(hex!("0000000000000000000000000000000000000005"));
const STORAGE: H256 =
H256(hex!("0000000000000000000000000000000000000000000000000000000000000001"));

Expand All @@ -217,27 +249,41 @@ mod tests {

tx.put::<tables::AccountHistory>(
ShardedKey { key: ADDRESS, highest_block_number: 7 },
BlockNumberList::new([3, 7]).unwrap(),
BlockNumberList::new([1, 3, 7]).unwrap(),
)
.unwrap();
tx.put::<tables::AccountHistory>(
ShardedKey { key: ADDRESS, highest_block_number: u64::MAX },
BlockNumberList::new([10, 15]).unwrap(),
)
.unwrap();
tx.put::<tables::AccountHistory>(
ShardedKey { key: HIGHER_ADDRESS, highest_block_number: u64::MAX },
BlockNumberList::new([4]).unwrap(),
)
.unwrap();

let acc_plain = Account { nonce: 100, balance: U256::ZERO, bytecode_hash: None };
let acc_at15 = Account { nonce: 15, balance: U256::ZERO, bytecode_hash: None };
let acc_at10 = Account { nonce: 10, balance: U256::ZERO, bytecode_hash: None };
let acc_at7 = Account { nonce: 7, balance: U256::ZERO, bytecode_hash: None };
let acc_at3 = Account { nonce: 3, balance: U256::ZERO, bytecode_hash: None };

let higher_acc_plain = Account { nonce: 4, balance: U256::ZERO, bytecode_hash: None };

// setup
tx.put::<tables::AccountChangeSet>(1, AccountBeforeTx { address: ADDRESS, info: None })
.unwrap();
tx.put::<tables::AccountChangeSet>(
3,
AccountBeforeTx { address: ADDRESS, info: Some(acc_at3) },
)
.unwrap();
tx.put::<tables::AccountChangeSet>(
4,
AccountBeforeTx { address: HIGHER_ADDRESS, info: None },
)
.unwrap();
tx.put::<tables::AccountChangeSet>(
7,
AccountBeforeTx { address: ADDRESS, info: Some(acc_at7) },
Expand All @@ -256,13 +302,15 @@ mod tests {

// setup plain state
tx.put::<tables::PlainAccountState>(ADDRESS, acc_plain).unwrap();
tx.put::<tables::PlainAccountState>(HIGHER_ADDRESS, higher_acc_plain).unwrap();
tx.commit().unwrap();

let tx = db.tx().unwrap();

// run
assert_eq!(HistoricalStateProviderRef::new(&tx, 1).basic_account(ADDRESS), Ok(None));
assert_eq!(
HistoricalStateProviderRef::new(&tx, 1).basic_account(ADDRESS),
HistoricalStateProviderRef::new(&tx, 2).basic_account(ADDRESS),
Ok(Some(acc_at3))
);
assert_eq!(
Expand Down Expand Up @@ -293,6 +341,12 @@ mod tests {
HistoricalStateProviderRef::new(&tx, 16).basic_account(ADDRESS),
Ok(Some(acc_plain))
);

assert_eq!(HistoricalStateProviderRef::new(&tx, 1).basic_account(HIGHER_ADDRESS), Ok(None));
assert_eq!(
HistoricalStateProviderRef::new(&tx, 1000).basic_account(HIGHER_ADDRESS),
Ok(Some(higher_acc_plain))
);
}

#[test]
Expand All @@ -316,7 +370,17 @@ mod tests {
BlockNumberList::new([10, 15]).unwrap(),
)
.unwrap();
tx.put::<tables::StorageHistory>(
StorageShardedKey {
address: HIGHER_ADDRESS,
sharded_key: ShardedKey { key: STORAGE, highest_block_number: u64::MAX },
},
BlockNumberList::new([4]).unwrap(),
)
.unwrap();

let higher_entry_plain = StorageEntry { key: STORAGE, value: U256::from(1000) };
let higher_entry_at4 = StorageEntry { key: STORAGE, value: U256::from(0) };
let entry_plain = StorageEntry { key: STORAGE, value: U256::from(100) };
let entry_at15 = StorageEntry { key: STORAGE, value: U256::from(15) };
let entry_at10 = StorageEntry { key: STORAGE, value: U256::from(10) };
Expand All @@ -325,25 +389,21 @@ mod tests {

// setup
tx.put::<tables::StorageChangeSet>((3, ADDRESS).into(), entry_at3).unwrap();
tx.put::<tables::StorageChangeSet>((4, HIGHER_ADDRESS).into(), higher_entry_at4).unwrap();
tx.put::<tables::StorageChangeSet>((7, ADDRESS).into(), entry_at7).unwrap();
tx.put::<tables::StorageChangeSet>((10, ADDRESS).into(), entry_at10).unwrap();
tx.put::<tables::StorageChangeSet>((15, ADDRESS).into(), entry_at15).unwrap();

// setup plain state
tx.put::<tables::PlainStorageState>(ADDRESS, entry_plain).unwrap();
tx.put::<tables::PlainStorageState>(HIGHER_ADDRESS, higher_entry_plain).unwrap();
tx.commit().unwrap();

let tx = db.tx().unwrap();

// run
assert_eq!(
HistoricalStateProviderRef::new(&tx, 0).storage(ADDRESS, STORAGE),
Ok(Some(entry_at3.value))
);
assert_eq!(
HistoricalStateProviderRef::new(&tx, 3).storage(ADDRESS, STORAGE),
Ok(Some(entry_at3.value))
);
assert_eq!(HistoricalStateProviderRef::new(&tx, 0).storage(ADDRESS, STORAGE), Ok(None));
assert_eq!(HistoricalStateProviderRef::new(&tx, 3).storage(ADDRESS, STORAGE), Ok(None));
assert_eq!(
HistoricalStateProviderRef::new(&tx, 4).storage(ADDRESS, STORAGE),
Ok(Some(entry_at7.value))
Expand All @@ -368,5 +428,13 @@ mod tests {
HistoricalStateProviderRef::new(&tx, 16).storage(ADDRESS, STORAGE),
Ok(Some(entry_plain.value))
);
assert_eq!(
HistoricalStateProviderRef::new(&tx, 1).storage(HIGHER_ADDRESS, STORAGE),
Ok(None)
);
assert_eq!(
HistoricalStateProviderRef::new(&tx, 1000).storage(HIGHER_ADDRESS, STORAGE),
Ok(Some(higher_entry_plain.value))
);
}
}