Skip to content

perf(drive-abci): stop rewriting the whole platform state every block #4571

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

Open
PastaPastaPasta wants to merge 1 commit into
base: perf/checkpoint-skip-during-replay
Choose a base branch
from
+447 −69
Open

perf(drive-abci): stop rewriting the whole platform state every block #4571

Show file tree
Hide file tree
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
5 changes: 5 additions & 0 deletions packages/rs-drive-abci/src/execution/platform_events/block_end/update_state_cache/v0/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,11 @@ where
// Persist block state
self.store_platform_state(&block_platform_state, Some(transaction), platform_version)?;

// Whatever the store wrote is now what is on disk for this block, so the
// next block only has to write the full record if it changes something
// heavy itself.
block_platform_state.heavy_fields_dirty = false;

let block_platform_state = Arc::new(block_platform_state);

self.state.store(block_platform_state);
Expand Down
17 changes: 17 additions & 0 deletions ...m_events/core_based_updates/update_masternode_list/update_state_masternode_list/v0/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,23 @@ where
..
} = &masternode_diff;

// Core advances a block without any masternode changing far more often
// than not. Returning before the first mutable borrow keeps the platform
// state clean, which is what lets the block skip rewriting the full saved
// state (over a megabyte on mainnet) to disk.
if !start_from_scratch
&& added_mns.is_empty()
&& removed_mns.is_empty()
&& updated_mns.is_empty()
{
return Ok(
update_state_masternode_list_outcome::v0::UpdateStateMasternodeListOutcome {
masternode_list_diff: masternode_diff,
removed_masternodes: BTreeMap::new(),
},
);
}

//todo: clean up
let added_hpmns = added_mns.iter().filter_map(|masternode| {
if masternode.node_type == MasternodeType::Evo {
Expand Down
109 changes: 69 additions & 40 deletions ...-drive-abci/src/execution/platform_events/core_based_updates/update_quorum_info/v0/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,27 +117,34 @@ where
.into_iter()
.collect();

let mut removed_a_validator_set = false;
// Checked before taking a mutable borrow: on most blocks Core reports the
// same quorums as the block before, and taking the borrow marks the whole
// platform state as needing a full rewrite to disk.
let removed_a_validator_set = block_platform_state
.validator_sets()
.keys()
.any(|quorum_hash| !validator_quorums_list.contains_key::<QuorumHash>(quorum_hash));

// Remove validator_sets entries that are no longer valid for the core block height
block_platform_state
.validator_sets_mut()
.retain(|quorum_hash, _| {
let retain = validator_quorums_list.contains_key::<QuorumHash>(quorum_hash);
removed_a_validator_set |= !retain;

if !retain {
tracing::trace!(
?quorum_hash,
quorum_type = ?self.config.validator_set.quorum_type,
"removed validator set {} with quorum type {}",
quorum_hash,
self.config.validator_set.quorum_type
)
}
if removed_a_validator_set {
block_platform_state
.validator_sets_mut()
.retain(|quorum_hash, _| {
let retain = validator_quorums_list.contains_key::<QuorumHash>(quorum_hash);

if !retain {
tracing::trace!(
?quorum_hash,
quorum_type = ?self.config.validator_set.quorum_type,
"removed validator set {} with quorum type {}",
quorum_hash,
self.config.validator_set.quorum_type
)
}

retain
});
retain
});
}

// Fetch quorum info and their keys from the RPC for new quorums
let mut quorum_infos = validator_quorums_list
Expand Down Expand Up @@ -192,25 +199,28 @@ where

let is_validator_set_updated = !new_validator_sets.is_empty() || removed_a_validator_set;

// Add new validator_sets entries
block_platform_state
.validator_sets_mut()
.extend(new_validator_sets);

// Sort all validator sets into deterministic order by core block height of creation
block_platform_state
.validator_sets_mut()
.sort_by(|_, quorum_a, _, quorum_b| {
let primary_comparison = quorum_b.core_height().cmp(&quorum_a.core_height());
if primary_comparison == std::cmp::Ordering::Equal {
quorum_b
.quorum_hash()
.cmp(quorum_a.quorum_hash())
.then_with(|| quorum_b.core_height().cmp(&quorum_a.core_height()))
} else {
primary_comparison
}
});
// Add new validator_sets entries. Nothing added and nothing removed means
// the map is already the one the previous block sorted, so leave it be.
if is_validator_set_updated {
block_platform_state
.validator_sets_mut()
.extend(new_validator_sets);

// Sort all validator sets into deterministic order by core block height of creation
block_platform_state
.validator_sets_mut()
.sort_by(|_, quorum_a, _, quorum_b| {
let primary_comparison = quorum_b.core_height().cmp(&quorum_a.core_height());
if primary_comparison == std::cmp::Ordering::Equal {
quorum_b
.quorum_hash()
.cmp(quorum_a.quorum_hash())
.then_with(|| quorum_b.core_height().cmp(&quorum_a.core_height()))
} else {
primary_comparison
}
});
}

// Update Chain Lock quorums

Expand All @@ -231,7 +241,7 @@ where
} else {
self.update_quorums_from_quorum_list(
quorum_set_type,
block_platform_state.chain_lock_validating_quorums_mut(),
block_platform_state,
platform_state,
&extended_quorum_list,
last_committed_core_height,
Expand Down Expand Up @@ -266,7 +276,7 @@ where
} else {
self.update_quorums_from_quorum_list(
quorum_set_type,
block_platform_state.instant_lock_validating_quorums_mut(),
block_platform_state,
platform_state,
&extended_quorum_list,
last_committed_core_height,
Expand Down Expand Up @@ -319,7 +329,7 @@ where
fn update_quorums_from_quorum_list(
&self,
quorum_set_type: QuorumSetType,
quorum_set: &mut SignatureVerificationQuorumSet,
block_platform_state: &mut PlatformState,
platform_state: Option<&PlatformState>,
full_quorum_list: &ExtendedQuorumListResult,
last_committed_core_height: u32,
Expand All @@ -341,6 +351,25 @@ where
})
.collect();

// Core reports the same quorums on most blocks. Decide read-only whether
// anything moved, because reaching for the mutable quorum set marks the
// whole platform state as needing a full rewrite to disk.
{
let current =
quorum_set_by_type(block_platform_state, &quorum_set_type).current_quorums();
let unchanged = current.len() == quorums_list.len()
&& current.iter().all(|(quorum_hash, quorum)| {
quorums_list
.get(quorum_hash)
.is_some_and(|index| *index == quorum.index)
});
if unchanged {
return Ok(false);
}
}

let quorum_set = quorum_set_by_type_mut(block_platform_state, &quorum_set_type);

let mut removed_a_validating_quorum = false;

// Remove validating_quorums entries that are no longer valid for the core block height
Expand Down
61 changes: 47 additions & 14 deletions packages/rs-drive-abci/src/execution/storage/fetch_platform_state/v0/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
use crate::error::Error;
use crate::platform_types::platform::Platform;
use crate::platform_types::platform_state::recent::PlatformStateRecent;
use crate::platform_types::platform_state::PlatformState;
use dpp::block::extended_block_info::v0::ExtendedBlockInfoV0Getters;
use dpp::serialization::PlatformDeserializableFromVersionedStructure;
use dpp::version::PlatformVersion;
use dpp::ProtocolError;
use drive::drive::Drive;
use drive::query::TransactionArg;

Expand All @@ -12,23 +15,53 @@ impl<C> Platform<C> {
transaction: TransactionArg,
platform_version: &PlatformVersion,
) -> Result<Option<PlatformState>, Error> {
drive
let Some(bytes) = drive
.fetch_platform_state_bytes(transaction, platform_version)
.map_err(Error::Drive)?
.map(|bytes| {
let result = PlatformState::versioned_deserialize(&bytes, platform_version)
.map_err(Error::Protocol);
else {
return Ok(None);
};

if result.is_err() {
tracing::error!(
bytes = hex::encode(&bytes),
"Unable deserialize platform state for version {}",
platform_version.protocol_version
);
}

result
let mut state = PlatformState::versioned_deserialize(&bytes, platform_version)
.inspect_err(|_| {
tracing::error!(
bytes = hex::encode(&bytes),
"Unable deserialize platform state for version {}",
platform_version.protocol_version
);
})
.transpose()
.map_err(Error::Protocol)?;

// The full record is only rewritten when a heavy field changes, so a
// newer small record holds the block info and quorum hashes for the
// blocks since. An older one (or none, on a database written before this
// existed) is ignored: the full record already has those fields.
if let Some(recent_bytes) = drive
.fetch_platform_state_recent_bytes(transaction)
.map_err(Error::Drive)?
{
let (recent, _): (PlatformStateRecent, _) = bincode::decode_from_slice(
&recent_bytes,
bincode::config::standard()
.with_big_endian()
.with_no_limit(),
)
.map_err(|e| {
Error::Protocol(ProtocolError::PlatformDeserializationError(format!(
"unable to deserialize recent platform state: {e}"
)))
})?;

if recent.height()
>= state
.last_committed_block_info
.as_ref()
.map(|i| i.basic_info().height)
{
recent.apply_to(&mut state);
}
}

Ok(Some(state))
}
}
59 changes: 48 additions & 11 deletions packages/rs-drive-abci/src/execution/storage/store_platform_state/v0/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
use crate::error::Error;
use crate::platform_types::platform::Platform;
use crate::platform_types::platform_state::recent::PlatformStateRecent;
use crate::platform_types::platform_state::PlatformState;
use dpp::block::extended_block_info::v0::ExtendedBlockInfoV0Getters;
use dpp::serialization::PlatformSerializable;
use dpp::version::PlatformVersion;
use dpp::ProtocolError;
use drive::query::TransactionArg;

impl<C> Platform<C> {
Expand All @@ -13,21 +16,55 @@ impl<C> Platform<C> {
platform_version: &PlatformVersion,
) -> Result<(), Error> {
#[cfg(feature = "testing-config")]
{
if self.config.testing_configs.store_platform_state {
let should_store = self.config.testing_configs.store_platform_state;
#[cfg(not(feature = "testing-config"))]
let should_store = true;

if should_store {
// The masternode lists, validator sets and quorum sets are most of
// the record — over a megabyte on mainnet — and only change when
// Core's do, which is a minority of blocks. While replaying history
// the full record is rewritten only when one of them moved, and the
// small record below carries the per-block fields in between. Both
// are written in the block's transaction, so a reader never sees
// them disagree.
//
// Once the node is at the tip the full record is written every block
// again, so a node that is up to date always has a complete record on
// disk and an older drive-abci — which knows nothing about the small
// record — can still read it. Skipping is confined to a node that is
// catching up, where the remedy for any format trouble is the resync
// it is already doing.
if state.heavy_fields_dirty
|| !state
.last_committed_block_info
.as_ref()
.is_some_and(|info| {
crate::utils::is_historical_block(info.basic_info().time_ms)
})
{
let bytes = state.serialize_to_bytes()?;
self.drive
.store_platform_state_bytes(
&state.serialize_to_bytes()?,
transaction,
platform_version,
)
.store_platform_state_bytes(&bytes, transaction, platform_version)
.map_err(Error::Drive)?;
}

let recent: PlatformStateRecent = state.into();
let recent_bytes = bincode::encode_to_vec(
recent,
bincode::config::standard()
.with_big_endian()
.with_no_limit(),
)
.map_err(|e| {
Error::Protocol(ProtocolError::PlatformSerializationError(format!(
"unable to serialize recent platform state: {e}"
)))
})?;
self.drive
.store_platform_state_recent_bytes(&recent_bytes, transaction)
.map_err(Error::Drive)?;
}
#[cfg(not(feature = "testing-config"))]
self.drive
.store_platform_state_bytes(&state.serialize_to_bytes()?, transaction, platform_version)
.map_err(Error::Drive)?;

// We need to persist new protocol version as well be able to read block state
self.drive
Expand Down
Loading
Loading