Skip to content
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.

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

Transaction writeset store #3903

Open
wants to merge 14 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 11 commits
Commits
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
86 changes: 76 additions & 10 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 3 additions & 0 deletions chain/api/src/message.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ use starcoin_types::{
transaction::Transaction,
};
use starcoin_vm_types::access_path::AccessPath;
use starcoin_vm_types::write_set::WriteSet;

#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug)]
Expand Down Expand Up @@ -60,6 +61,7 @@ pub enum ChainRequest {
access_path: Option<AccessPath>,
},
GetBlockInfos(Vec<HashValue>),
GetTransactionWriteSet(HashValue),
}

impl ServiceRequest for ChainRequest {
Expand Down Expand Up @@ -88,4 +90,5 @@ pub enum ChainResponse {
HashVec(Vec<HashValue>),
TransactionProof(Box<Option<TransactionInfoWithProof>>),
BlockInfoVec(Box<Vec<Option<BlockInfo>>>),
TransactionWriteSet(Option<WriteSet>),
}
16 changes: 16 additions & 0 deletions chain/api/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ use starcoin_types::{
startup_info::StartupInfo,
};
use starcoin_vm_types::access_path::AccessPath;
use starcoin_vm_types::write_set::WriteSet;

/// Readable block chain service trait
pub trait ReadableChainService {
Expand Down Expand Up @@ -72,6 +73,8 @@ pub trait ReadableChainService {
) -> Result<Option<TransactionInfoWithProof>>;

fn get_block_infos(&self, ids: Vec<HashValue>) -> Result<Vec<Option<BlockInfo>>>;

fn get_transaction_write_set(&self, hash: HashValue) -> Result<Option<WriteSet>>;
}

/// Writeable block chain service trait
Expand Down Expand Up @@ -139,6 +142,8 @@ pub trait ChainAsyncService:
) -> Result<Option<TransactionInfoWithProof>>;

async fn get_block_infos(&self, hashes: Vec<HashValue>) -> Result<Vec<Option<BlockInfo>>>;

async fn get_transaction_write_set(&self, hash: HashValue) -> Result<Option<WriteSet>>;
}

#[async_trait::async_trait]
Expand Down Expand Up @@ -436,4 +441,15 @@ where
bail!("get block_infos error")
}
}

async fn get_transaction_write_set(&self, hash: HashValue) -> Result<Option<WriteSet>> {
let response = self
.send(ChainRequest::GetTransactionWriteSet(hash))
.await??;
if let ChainResponse::TransactionWriteSet(write_set) = response {
Ok(write_set)
} else {
bail!("get get_write_set error")
}
}
}
8 changes: 8 additions & 0 deletions chain/service/src/chain_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ use starcoin_types::{
};
use starcoin_vm_runtime::metrics::VMMetrics;
use starcoin_vm_types::access_path::AccessPath;
use starcoin_vm_types::write_set::WriteSet;
use std::sync::Arc;

/// A Chain reader service to provider Reader API.
Expand Down Expand Up @@ -232,6 +233,9 @@ impl ServiceHandler<Self, ChainRequest> for ChainReaderService {
ChainRequest::GetBlockInfos(ids) => Ok(ChainResponse::BlockInfoVec(Box::new(
self.inner.get_block_infos(ids)?,
))),
ChainRequest::GetTransactionWriteSet(hash) => Ok(ChainResponse::TransactionWriteSet(
self.inner.get_transaction_write_set(hash)?,
)),
}
}
}
Expand Down Expand Up @@ -416,6 +420,10 @@ impl ReadableChainService for ChainReaderServiceInner {
fn get_block_infos(&self, ids: Vec<HashValue>) -> Result<Vec<Option<BlockInfo>>> {
self.storage.get_block_infos(ids)
}

fn get_transaction_write_set(&self, hash: HashValue) -> Result<Option<WriteSet>> {
self.storage.get_write_set(hash)
}
}

#[cfg(test)]
Expand Down
6 changes: 6 additions & 0 deletions chain/src/chain.rs
Original file line number Diff line number Diff line change
Expand Up @@ -463,6 +463,7 @@ impl BlockChain {
let block_id = block.id();
let txn_infos = executed_data.txn_infos;
let txn_events = executed_data.txn_events;
let txn_write_set = executed_data.txn_write_sets;

debug_assert!(
txn_events.len() == txn_infos.len(),
Expand Down Expand Up @@ -505,6 +506,11 @@ impl BlockChain {

storage.save_block_info(block_info.clone())?;

for (hash_value, write_set) in txn_write_set{
storage.save_write_set(hash_value, write_set)?;
Copy link
Collaborator

@nkysg nkysg Jun 28, 2023

Choose a reason for hiding this comment

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

看起来 storage.save_write_set可以弄成一个batch_write,你参考下save_transaction_batch看看是不是这样, batch_write主要是少了几次rocksdb调用,这个调用虽然比较小,再就是现在是一次性写入,现在是rocksdb异步刷盘,不是每次都flush磁盘,如果是同步就会每次都flush磁盘就很慢了. @welbon @simonjiao

Copy link
Contributor Author

Choose a reason for hiding this comment

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

这里考虑上线之前加一下

Copy link
Contributor Author

Choose a reason for hiding this comment

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

这里加了batch处理

}


watch(CHAIN_WATCH_NAME, "n26");
Ok(ExecutedBlock { block, block_info })
}
Expand Down
10 changes: 9 additions & 1 deletion executor/src/block_executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,14 @@ use starcoin_types::transaction::TransactionStatus;
use starcoin_types::transaction::{Transaction, TransactionInfo};
use starcoin_vm_runtime::metrics::VMMetrics;
use starcoin_vm_types::contract_event::ContractEvent;
use starcoin_vm_types::write_set::WriteSet;

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BlockExecutedData {
pub state_root: HashValue,
pub txn_infos: Vec<TransactionInfo>,
pub txn_events: Vec<Vec<ContractEvent>>,
pub txn_write_sets: Vec<(HashValue, WriteSet)>,
}

impl Default for BlockExecutedData {
Expand All @@ -23,6 +25,7 @@ impl Default for BlockExecutedData {
state_root: HashValue::zero(),
txn_events: vec![],
txn_infos: vec![],
txn_write_sets: vec![],
}
}
}
Expand Down Expand Up @@ -53,12 +56,13 @@ pub fn block_execute<S: ChainStateReader + ChainStateWriter>(
}
TransactionStatus::Keep(status) => {
chain_state
.apply_write_set(write_set)
.apply_write_set(write_set.clone())
.map_err(BlockExecutorError::BlockChainStateErr)?;

let txn_state_root = chain_state
.commit()
.map_err(BlockExecutorError::BlockChainStateErr)?;

#[cfg(testing)]
info!("txn_hash {} gas_used {}", txn_hash, gas_used);
executed_data.txn_infos.push(TransactionInfo::new(
Expand All @@ -68,7 +72,11 @@ pub fn block_execute<S: ChainStateReader + ChainStateWriter>(
gas_used,
status,
));

executed_data.txn_events.push(events);

// Put write set into result
executed_data.txn_write_sets.push((txn_hash, write_set));
}
};
}
Expand Down
Loading