Skip to content

Commit 195bdb3

Browse files
wqfishwqfish
committed
hot state e2e
1 parent c0fb41f commit 195bdb3

File tree

27 files changed

+879
-364
lines changed

27 files changed

+879
-364
lines changed

‎config/src/config/storage_config.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -358,7 +358,7 @@ pub struct PrunerConfig {
358358
impl Default for LedgerPrunerConfig {
359359
fn default() -> Self {
360360
LedgerPrunerConfig {
361-
enable: true,
361+
enable: false,
362362
prune_window: 90_000_000,
363363
batch_size: 5_000,
364364
user_pruning_window_offset: 200_000,
@@ -369,7 +369,7 @@ impl Default for LedgerPrunerConfig {
369369
impl Default for StateMerklePrunerConfig {
370370
fn default() -> Self {
371371
StateMerklePrunerConfig {
372-
enable: true,
372+
enable: false,
373373
// This allows a block / chunk being executed to have access to a non-latest state tree.
374374
// It needs to be greater than the number of versions the state committing thread is
375375
// able to commit during the execution of the block / chunk. If the bad case indeed
@@ -386,7 +386,7 @@ impl Default for StateMerklePrunerConfig {
386386
impl Default for EpochSnapshotPrunerConfig {
387387
fn default() -> Self {
388388
Self {
389-
enable: true,
389+
enable: false,
390390
// This is based on ~5K TPS * 2h/epoch * 2 epochs. -- epoch ending snapshots are used
391391
// by state sync in fast sync mode.
392392
// The setting is in versions, not epochs, because this makes it behave more like other

‎execution/executor-types/src/execution_output.rs‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,15 @@ use aptos_storage_interface::state_store::{
1414
use aptos_types::{
1515
contract_event::ContractEvent,
1616
epoch_state::EpochState,
17-
state_store::hot_state::HotStateConfig,
17+
state_store::{
18+
hot_state::HotStateConfig, state_key::StateKey, state_value::StateValue, NUM_STATE_SHARDS,
19+
},
1820
transaction::{
1921
block_epilogue::BlockEndInfo, ExecutionStatus, Transaction, TransactionStatus, Version,
2022
},
2123
};
2224
use derive_more::Deref;
23-
use std::sync::Arc;
25+
use std::{collections::HashMap, sync::Arc};
2426

2527
#[derive(Clone, Debug, Deref)]
2628
pub struct ExecutionOutput {
@@ -41,6 +43,8 @@ impl ExecutionOutput {
4143
block_end_info: Option<BlockEndInfo>,
4244
next_epoch_state: Option<EpochState>,
4345
subscribable_events: Planned<Vec<ContractEvent>>,
46+
hot_inserted: [HashMap<StateKey, Option<StateValue>>; NUM_STATE_SHARDS],
47+
hot_evicted: [HashMap<StateKey, Option<StateValue>>; NUM_STATE_SHARDS],
4448
) -> Self {
4549
let next_version = first_version + to_commit.len() as Version;
4650
assert_eq!(next_version, result_state.latest().next_version());
@@ -65,6 +69,8 @@ impl ExecutionOutput {
6569
block_end_info,
6670
next_epoch_state,
6771
subscribable_events,
72+
hot_inserted,
73+
hot_evicted,
6874
})
6975
}
7076

@@ -81,6 +87,8 @@ impl ExecutionOutput {
8187
block_end_info: None,
8288
next_epoch_state: None,
8389
subscribable_events: Planned::ready(vec![]),
90+
hot_inserted: std::array::from_fn(|_| HashMap::new()),
91+
hot_evicted: std::array::from_fn(|_| HashMap::new()),
8492
})
8593
}
8694

@@ -99,6 +107,8 @@ impl ExecutionOutput {
99107
block_end_info: None,
100108
next_epoch_state: None,
101109
subscribable_events: Planned::ready(vec![]),
110+
hot_inserted: std::array::from_fn(|_| HashMap::new()),
111+
hot_evicted: std::array::from_fn(|_| HashMap::new()),
102112
})
103113
}
104114

@@ -119,6 +129,8 @@ impl ExecutionOutput {
119129
block_end_info: None,
120130
next_epoch_state: self.next_epoch_state.clone(),
121131
subscribable_events: Planned::ready(vec![]),
132+
hot_inserted: std::array::from_fn(|_| HashMap::new()),
133+
hot_evicted: std::array::from_fn(|_| HashMap::new()),
122134
})
123135
}
124136

@@ -166,6 +178,8 @@ pub struct Inner {
166178
/// state cache.
167179
pub next_epoch_state: Option<EpochState>,
168180
pub subscribable_events: Planned<Vec<ContractEvent>>,
181+
pub hot_inserted: [HashMap<StateKey, Option<StateValue>>; NUM_STATE_SHARDS],
182+
pub hot_evicted: [HashMap<StateKey, Option<StateValue>>; NUM_STATE_SHARDS],
169183
}
170184

171185
impl Inner {

‎execution/executor/src/block_executor/mod.rs‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -315,7 +315,14 @@ where
315315
output.set_state_checkpoint_output(DoStateCheckpoint::run(
316316
&output.execution_output,
317317
parent_block.output.ensure_result_state_summary()?,
318-
&ProvableStateSummary::new_persisted(self.db.reader.as_ref())?,
318+
&ProvableStateSummary::new_persisted(
319+
self.db.reader.as_ref(),
320+
/* is_hot = */ true,
321+
)?,
322+
&ProvableStateSummary::new_persisted(
323+
self.db.reader.as_ref(),
324+
/* is_hot = */ false,
325+
)?,
319326
None,
320327
)?);
321328
output.set_ledger_update_output(DoLedgerUpdate::run(

‎execution/executor/src/chunk_executor/mod.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -346,7 +346,8 @@ impl<V: VMBlockExecutor> ChunkExecutorInner<V> {
346346
let state_checkpoint_output = DoStateCheckpoint::run(
347347
&output.execution_output,
348348
&parent_state_summary,
349-
&ProvableStateSummary::new_persisted(self.db.reader.as_ref())?,
349+
&ProvableStateSummary::new_persisted(self.db.reader.as_ref(), true)?,
350+
&ProvableStateSummary::new_persisted(self.db.reader.as_ref(), false)?,
350351
Some(
351352
chunk_verifier
352353
.transaction_infos()

‎execution/executor/src/workflow/do_get_execution_output.rs‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ use aptos_types::{
5151
};
5252
use aptos_vm::VMBlockExecutor;
5353
use itertools::Itertools;
54-
use std::sync::Arc;
54+
use std::{collections::HashMap, sync::Arc};
5555

5656
pub struct DoGetExecutionOutput;
5757

@@ -171,7 +171,7 @@ impl DoGetExecutionOutput {
171171
}
172172
}
173173

174-
Parser::parse(
174+
let parsed = Parser::parse(
175175
state_view.next_version(),
176176
transactions,
177177
transaction_outputs,
@@ -182,7 +182,9 @@ impl DoGetExecutionOutput {
182182
transaction_slice_metadata
183183
.append_state_checkpoint_to_block()
184184
.is_some(),
185-
)
185+
)?;
186+
info!("Finished executing transactions");
187+
Ok(parsed)
186188
}
187189

188190
pub fn by_transaction_execution_sharded<V: VMBlockExecutor>(
@@ -417,12 +419,18 @@ impl Parser {
417419
},
418420
)?;
419421

420-
let result_state = parent_state.update_with_memorized_reads(
422+
let (result_state, hot_inserted, hot_evicted) = parent_state.update_with_memorized_reads(
421423
base_state_view.persisted_hot_state(),
422424
base_state_view.persisted_state(),
423425
to_commit.state_update_refs(),
424426
base_state_view.memorized_reads(),
425427
);
428+
// for (i, shard) in hot_inserted.iter().enumerate() {
429+
// info!("shard {}: # inserted: {}", i, shard.len());
430+
// }
431+
// for (i, shard) in hot_evicted.iter().enumerate() {
432+
// info!("shard {}: # evicted: {}", i, shard.len());
433+
// }
426434
let state_reads = base_state_view.into_memorized_reads();
427435

428436
let out = ExecutionOutput::new(
@@ -437,6 +445,8 @@ impl Parser {
437445
block_end_info,
438446
next_epoch_state,
439447
Planned::place_holder(),
448+
hot_inserted,
449+
hot_evicted,
440450
);
441451
let ret = out.clone();
442452
ret.subscribable_events

‎execution/executor/src/workflow/do_state_checkpoint.rs‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ use aptos_crypto::HashValue;
77
use aptos_executor_types::{
88
execution_output::ExecutionOutput, state_checkpoint_output::StateCheckpointOutput,
99
};
10+
use aptos_logger::info;
1011
use aptos_metrics_core::TimerHelper;
1112
use aptos_storage_interface::state_store::state_summary::{
1213
LedgerStateSummary, ProvableStateSummary,
@@ -18,14 +19,19 @@ impl DoStateCheckpoint {
1819
pub fn run(
1920
execution_output: &ExecutionOutput,
2021
parent_state_summary: &LedgerStateSummary,
21-
persisted_state_summary: &ProvableStateSummary,
22+
hot_persisted_state_summary: &ProvableStateSummary,
23+
cold_persisted_state_summary: &ProvableStateSummary,
2224
known_state_checkpoints: Option<Vec<Option<HashValue>>>,
2325
) -> Result<StateCheckpointOutput> {
2426
let _timer = OTHER_TIMERS.timer_with(&["do_state_checkpoint"]);
27+
// info!("Start getting state checkpoint");
2528

2629
let state_summary = parent_state_summary.update(
27-
persisted_state_summary,
28-
execution_output.to_commit.state_update_refs(),
30+
hot_persisted_state_summary,
31+
cold_persisted_state_summary,
32+
&execution_output.hot_inserted,
33+
&execution_output.hot_evicted,
34+
execution_output.next_version(),
2935
)?;
3036

3137
let state_checkpoint_hashes = Self::get_state_checkpoint_hashes(
@@ -34,6 +40,8 @@ impl DoStateCheckpoint {
3440
&state_summary,
3541
)?;
3642

43+
// info!("Finished getting state checkpoint");
44+
3745
Ok(StateCheckpointOutput::new(
3846
state_summary,
3947
state_checkpoint_hashes,

‎execution/executor/src/workflow/mod.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,8 @@ impl ApplyExecutionOutput {
2727
let state_checkpoint_output = DoStateCheckpoint::run(
2828
&execution_output,
2929
&base_view.state_summary,
30-
&ProvableStateSummary::new_persisted(reader)?,
30+
&ProvableStateSummary::new_persisted(reader, true)?,
31+
&ProvableStateSummary::new_persisted(reader, false)?,
3132
None,
3233
)?;
3334
let ledger_update_output = DoLedgerUpdate::run(

‎storage/aptosdb/src/db/aptosdb_reader.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -660,12 +660,13 @@ impl DbReader for AptosDB {
660660
key_hash: &HashValue,
661661
version: Version,
662662
root_depth: usize,
663+
is_hot: bool,
663664
) -> Result<SparseMerkleProofExt> {
664665
gauged_api("get_state_proof_by_version_ext", || {
665666
self.error_if_state_merkle_pruned("State merkle", version)?;
666667

667668
self.state_store
668-
.get_state_proof_by_version_ext(key_hash, version, root_depth)
669+
.get_state_proof_by_version_ext(key_hash, version, root_depth, is_hot)
669670
})
670671
}
671672

@@ -674,12 +675,13 @@ impl DbReader for AptosDB {
674675
key_hash: &HashValue,
675676
version: Version,
676677
root_depth: usize,
678+
is_hot: bool,
677679
) -> Result<(Option<StateValue>, SparseMerkleProofExt)> {
678680
gauged_api("get_state_value_with_proof_by_version_ext", || {
679681
self.error_if_state_merkle_pruned("State merkle", version)?;
680682

681683
self.state_store
682-
.get_state_value_with_proof_by_version_ext(key_hash, version, root_depth)
684+
.get_state_value_with_proof_by_version_ext(key_hash, version, root_depth, is_hot)
683685
})
684686
}
685687

‎storage/aptosdb/src/db/aptosdb_testonly.rs‎

Lines changed: 59 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -130,67 +130,68 @@ impl AptosDB {
130130
ledger_info_with_sigs: Option<&LedgerInfoWithSignatures>,
131131
sync_commit: bool,
132132
) -> Result<()> {
133-
let (transactions, transaction_outputs, transaction_infos) =
134-
Self::disassemble_txns_to_commit(txns_to_commit);
135-
// Keep auxiliary info consistent with what was used to create TransactionInfo
136-
// Use block-relative indices to match what was used during TransactionInfo creation
137-
let persisted_auxiliary_infos = txns_to_commit
138-
.iter()
139-
.map(
140-
|txn_to_commit| match txn_to_commit.transaction_info.auxiliary_info_hash() {
141-
Some(hash) => {
142-
for i in 0..100 {
143-
if hash
144-
== CryptoHash::hash(&PersistedAuxiliaryInfo::V1 {
145-
transaction_index: i as u32,
146-
})
147-
{
148-
return PersistedAuxiliaryInfo::V1 {
149-
transaction_index: i as u32,
150-
};
151-
}
152-
}
153-
panic!("Hash not found");
154-
},
155-
None => PersistedAuxiliaryInfo::None,
156-
},
157-
)
158-
.collect();
159-
let transactions_to_keep = TransactionsToKeep::make(
160-
first_version,
161-
transactions,
162-
transaction_outputs,
163-
persisted_auxiliary_infos,
164-
);
133+
unimplemented!();
134+
// let (transactions, transaction_outputs, transaction_infos) =
135+
// Self::disassemble_txns_to_commit(txns_to_commit);
136+
// // Keep auxiliary info consistent with what was used to create TransactionInfo
137+
// // Use block-relative indices to match what was used during TransactionInfo creation
138+
// let persisted_auxiliary_infos = txns_to_commit
139+
// .iter()
140+
// .map(
141+
// |txn_to_commit| match txn_to_commit.transaction_info.auxiliary_info_hash() {
142+
// Some(hash) => {
143+
// for i in 0..100 {
144+
// if hash
145+
// == CryptoHash::hash(&PersistedAuxiliaryInfo::V1 {
146+
// transaction_index: i as u32,
147+
// })
148+
// {
149+
// return PersistedAuxiliaryInfo::V1 {
150+
// transaction_index: i as u32,
151+
// };
152+
// }
153+
// }
154+
// panic!("Hash not found");
155+
// },
156+
// None => PersistedAuxiliaryInfo::None,
157+
// },
158+
// )
159+
// .collect();
160+
// let transactions_to_keep = TransactionsToKeep::make(
161+
// first_version,
162+
// transactions,
163+
// transaction_outputs,
164+
// persisted_auxiliary_infos,
165+
// );
165166

166-
let current = self.state_store.current_state_locked().clone();
167-
let (hot_state, persisted_state) = self.state_store.get_persisted_state()?;
168-
let (new_state, reads) = current.ledger_state().update_with_db_reader(
169-
&persisted_state,
170-
hot_state,
171-
transactions_to_keep.state_update_refs(),
172-
self.state_store.clone(),
173-
)?;
174-
let persisted_summary = self.state_store.get_persisted_state_summary()?;
175-
let new_state_summary = current.ledger_state_summary().update(
176-
&ProvableStateSummary::new(persisted_summary, self),
177-
transactions_to_keep.state_update_refs(),
178-
)?;
167+
// let current = self.state_store.current_state_locked().clone();
168+
// let (hot_state, persisted_state) = self.state_store.get_persisted_state()?;
169+
// let (new_state, reads) = current.ledger_state().update_with_db_reader(
170+
// &persisted_state,
171+
// hot_state,
172+
// transactions_to_keep.state_update_refs(),
173+
// self.state_store.clone(),
174+
// )?;
175+
// let persisted_summary = self.state_store.get_persisted_state_summary()?;
176+
// let new_state_summary = current.ledger_state_summary().update(
177+
// &ProvableStateSummary::new(persisted_summary, self),
178+
// transactions_to_keep.state_update_refs(),
179+
// )?;
179180

180-
let chunk = ChunkToCommit {
181-
first_version,
182-
transactions: &transactions_to_keep.transactions,
183-
persisted_auxiliary_infos: &transactions_to_keep.persisted_auxiliary_infos,
184-
transaction_outputs: &transactions_to_keep.transaction_outputs,
185-
transaction_infos: &transaction_infos,
186-
state: &new_state,
187-
state_summary: &new_state_summary,
188-
state_update_refs: transactions_to_keep.state_update_refs(),
189-
state_reads: &reads,
190-
is_reconfig: transactions_to_keep.is_reconfig(),
191-
};
181+
// let chunk = ChunkToCommit {
182+
// first_version,
183+
// transactions: &transactions_to_keep.transactions,
184+
// persisted_auxiliary_infos: &transactions_to_keep.persisted_auxiliary_infos,
185+
// transaction_outputs: &transactions_to_keep.transaction_outputs,
186+
// transaction_infos: &transaction_infos,
187+
// state: &new_state,
188+
// state_summary: &new_state_summary,
189+
// state_update_refs: transactions_to_keep.state_update_refs(),
190+
// state_reads: &reads,
191+
// is_reconfig: transactions_to_keep.is_reconfig(),
192+
// };
192193

193-
self.save_transactions(chunk, ledger_info_with_sigs, sync_commit)
194+
// self.save_transactions(chunk, ledger_info_with_sigs, sync_commit)
194195
}
195196

196197
fn disassemble_txns_to_commit(

‎storage/aptosdb/src/db/test_helper.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -432,7 +432,7 @@ fn verify_snapshots(
432432
);
433433
for (state_key, state_value) in &updates {
434434
let (state_value_in_db, proof) = db
435-
.get_state_value_with_proof_by_version(state_key, snapshot_version)
435+
.get_state_value_with_proof_by_version(state_key, snapshot_version, false)
436436
.unwrap();
437437
assert_eq!(state_value_in_db.as_ref(), state_value.as_ref());
438438
proof

0 commit comments

Comments
 (0)