support async padding

This commit is contained in:
Peter Zhang 2024-09-14 00:11:31 +08:00
parent 59d24b073d
commit 814bc35b3b
9 changed files with 99 additions and 39 deletions

1
Cargo.lock generated
View File

@ -7243,6 +7243,7 @@ dependencies = [
"static_assertions", "static_assertions",
"tempdir", "tempdir",
"tiny-keccak", "tiny-keccak",
"tokio",
"tracing", "tracing",
"typenum", "typenum",
"zgs_seal", "zgs_seal",

View File

@ -423,7 +423,7 @@ impl Libp2pEventHandler {
let addr = self.get_listen_addr_or_add().await?; let addr = self.get_listen_addr_or_add().await?;
let timestamp = timestamp_now(); let timestamp = timestamp_now();
let shard_config = self.store.get_store().flow().get_shard_config(); let shard_config = self.store.get_store().get_shard_config();
let msg = AnnounceFile { let msg = AnnounceFile {
tx_ids, tx_ids,
@ -699,7 +699,7 @@ impl Libp2pEventHandler {
} }
// notify sync layer if shard config matches // notify sync layer if shard config matches
let my_shard_config = self.store.get_store().flow().get_shard_config(); let my_shard_config = self.store.get_store().get_shard_config();
if my_shard_config.intersect(&announced_shard_config) { if my_shard_config.intersect(&announced_shard_config) {
for tx_id in msg.tx_ids.iter() { for tx_id in msg.tx_ids.iter() {
self.send_to_sync(SyncMessage::AnnounceFileGossip { self.send_to_sync(SyncMessage::AnnounceFileGossip {

View File

@ -180,7 +180,7 @@ impl RpcServer for RpcServerImpl {
async fn get_shard_config(&self) -> RpcResult<ShardConfig> { async fn get_shard_config(&self) -> RpcResult<ShardConfig> {
debug!("zgs_getShardConfig"); debug!("zgs_getShardConfig");
let shard_config = self.ctx.log_store.get_store().flow().get_shard_config(); let shard_config = self.ctx.log_store.get_store().get_shard_config();
Ok(shard_config) Ok(shard_config)
} }

View File

@ -95,22 +95,22 @@ impl Store {
&self, &self,
seal_index_max: usize, seal_index_max: usize,
) -> anyhow::Result<Option<Vec<SealTask>>> { ) -> anyhow::Result<Option<Vec<SealTask>>> {
self.spawn(move |store| store.flow().pull_seal_chunk(seal_index_max)) self.spawn(move |store| store.pull_seal_chunk(seal_index_max))
.await .await
} }
pub async fn submit_seal_result(&self, answers: Vec<SealAnswer>) -> anyhow::Result<()> { pub async fn submit_seal_result(&self, answers: Vec<SealAnswer>) -> anyhow::Result<()> {
self.spawn(move |store| store.flow().submit_seal_result(answers)) self.spawn(move |store| store.submit_seal_result(answers))
.await .await
} }
pub async fn load_sealed_data(&self, chunk_index: u64) -> Result<Option<MineLoadChunk>> { pub async fn load_sealed_data(&self, chunk_index: u64) -> Result<Option<MineLoadChunk>> {
self.spawn(move |store| store.flow().load_sealed_data(chunk_index)) self.spawn(move |store| store.load_sealed_data(chunk_index))
.await .await
} }
pub async fn get_num_entries(&self) -> Result<u64> { pub async fn get_num_entries(&self) -> Result<u64> {
self.spawn(move |store| store.flow().get_num_entries()) self.spawn(move |store| store.get_num_entries())
.await .await
} }
@ -122,7 +122,7 @@ impl Store {
pub async fn update_shard_config(&self, shard_config: ShardConfig) { pub async fn update_shard_config(&self, shard_config: ShardConfig) {
self.spawn(move |store| { self.spawn(move |store| {
store.flow().update_shard_config(shard_config); store.update_shard_config(shard_config);
Ok(()) Ok(())
}) })
.await .await

View File

@ -29,6 +29,7 @@ itertools = "0.13.0"
serde = { version = "1.0.197", features = ["derive"] } serde = { version = "1.0.197", features = ["derive"] }
parking_lot = "0.12.3" parking_lot = "0.12.3"
serde_json = "1.0.127" serde_json = "1.0.127"
tokio = { version = "1.10.0", features = ["sync"] }
[dev-dependencies] [dev-dependencies]
tempdir = "0.3.7" tempdir = "0.3.7"

View File

@ -1,3 +1,4 @@
use crate::config::ShardConfig;
use crate::log_store::flow_store::{batch_iter_sharded, FlowConfig, FlowStore}; use crate::log_store::flow_store::{batch_iter_sharded, FlowConfig, FlowStore};
use crate::log_store::tx_store::TransactionStore; use crate::log_store::tx_store::TransactionStore;
use crate::log_store::{ use crate::log_store::{
@ -21,10 +22,12 @@ use std::cmp::Ordering;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::path::Path; use std::path::Path;
use std::sync::Arc; use std::sync::Arc;
use std::thread;
use std::sync::mpsc;
use tracing::{debug, error, info, instrument, trace, warn}; use tracing::{debug, error, info, instrument, trace, warn};
use super::tx_store::BlockHashAndSubmissionIndex; use super::tx_store::BlockHashAndSubmissionIndex;
use super::LogStoreInner; use super::{FlowSeal, MineLoadChunk, SealAnswer, SealTask};
/// 256 Bytes /// 256 Bytes
pub const ENTRY_SIZE: usize = 256; pub const ENTRY_SIZE: usize = 256;
@ -45,11 +48,18 @@ pub const COL_NUM: u32 = 9;
// Process at most 1M entries (256MB) pad data at a time. // Process at most 1M entries (256MB) pad data at a time.
const PAD_MAX_SIZE: usize = 1 << 20; const PAD_MAX_SIZE: usize = 1 << 20;
pub struct UpdateFlowMessage {
pub root_map: BTreeMap<usize, (H256, usize)>,
pub pad_data: Vec<u8>,
pub tx_start_flow_index: u64,
}
pub struct LogManager { pub struct LogManager {
pub(crate) db: Arc<dyn ZgsKeyValueDB>, pub(crate) db: Arc<dyn ZgsKeyValueDB>,
tx_store: TransactionStore, tx_store: TransactionStore,
flow_store: FlowStore, flow_store: Arc<FlowStore>,
merkle: RwLock<MerkleManager>, merkle: RwLock<MerkleManager>,
sender: mpsc::Sender<UpdateFlowMessage>,
} }
struct MerkleManager { struct MerkleManager {
@ -139,16 +149,6 @@ pub struct LogConfig {
pub flow: FlowConfig, pub flow: FlowConfig,
} }
impl LogStoreInner for LogManager {
fn flow(&self) -> &dyn super::Flow {
&self.flow_store
}
fn flow_mut(&mut self) -> &mut dyn super::Flow {
&mut self.flow_store
}
}
impl LogStoreChunkWrite for LogManager { impl LogStoreChunkWrite for LogManager {
fn put_chunks(&self, tx_seq: u64, chunks: ChunkArray) -> Result<()> { fn put_chunks(&self, tx_seq: u64, chunks: ChunkArray) -> Result<()> {
let mut merkle = self.merkle.write(); let mut merkle = self.merkle.write();
@ -389,6 +389,14 @@ impl LogStoreWrite for LogManager {
fn delete_block_hash_by_number(&self, block_number: u64) -> Result<()> { fn delete_block_hash_by_number(&self, block_number: u64) -> Result<()> {
self.tx_store.delete_block_hash_by_number(block_number) self.tx_store.delete_block_hash_by_number(block_number)
} }
fn update_shard_config(&self, shard_config: ShardConfig) {
self.flow_store.update_shard_config(shard_config)
}
fn submit_seal_result(&self, answers: Vec<SealAnswer>) -> Result<()> {
self.flow_store.submit_seal_result(answers)
}
} }
impl LogStoreChunkRead for LogManager { impl LogStoreChunkRead for LogManager {
@ -579,6 +587,22 @@ impl LogStoreRead for LogManager {
fn check_tx_pruned(&self, tx_seq: u64) -> crate::error::Result<bool> { fn check_tx_pruned(&self, tx_seq: u64) -> crate::error::Result<bool> {
self.tx_store.check_tx_pruned(tx_seq) self.tx_store.check_tx_pruned(tx_seq)
} }
fn pull_seal_chunk(&self, seal_index_max: usize) -> Result<Option<Vec<SealTask>>> {
self.flow_store.pull_seal_chunk(seal_index_max)
}
fn get_num_entries(&self) -> Result<u64> {
self.flow_store.get_num_entries()
}
fn load_sealed_data(&self, chunk_index: u64) -> Result<Option<MineLoadChunk>> {
self.flow_store.load_sealed_data(chunk_index)
}
fn get_shard_config(&self) -> ShardConfig {
self.flow_store.get_shard_config()
}
} }
impl LogManager { impl LogManager {
@ -596,7 +620,7 @@ impl LogManager {
fn new(db: Arc<dyn ZgsKeyValueDB>, config: LogConfig) -> Result<Self> { fn new(db: Arc<dyn ZgsKeyValueDB>, config: LogConfig) -> Result<Self> {
let tx_store = TransactionStore::new(db.clone())?; let tx_store = TransactionStore::new(db.clone())?;
let flow_store = FlowStore::new(db.clone(), config.flow); let flow_store = Arc::new(FlowStore::new(db.clone(), config.flow));
let mut initial_data = flow_store.get_chunk_root_list()?; let mut initial_data = flow_store.get_chunk_root_list()?;
// If the last tx `put_tx` does not complete, we will revert it in `initial_data.subtree_list` // If the last tx `put_tx` does not complete, we will revert it in `initial_data.subtree_list`
// first and call `put_tx` later. The known leaves in its data will be saved in `extra_leaves` // first and call `put_tx` later. The known leaves in its data will be saved in `extra_leaves`
@ -700,13 +724,19 @@ impl LogManager {
pora_chunks_merkle, pora_chunks_merkle,
last_chunk_merkle, last_chunk_merkle,
}); });
let log_manager = Self {
let (sender, receiver) = mpsc::channel();
let mut log_manager = Self {
db, db,
tx_store, tx_store,
flow_store, flow_store,
merkle, merkle,
sender
}; };
log_manager.start_receiver(receiver);
if let Some(tx) = last_tx_to_insert { if let Some(tx) = last_tx_to_insert {
log_manager.put_tx(tx)?; log_manager.put_tx(tx)?;
let mut merkle = log_manager.merkle.write(); let mut merkle = log_manager.merkle.write();
@ -727,6 +757,30 @@ impl LogManager {
Ok(log_manager) Ok(log_manager)
} }
fn start_receiver(&mut self, rx: mpsc::Receiver<UpdateFlowMessage>) {
let flow_store = self.flow_store.clone();
thread::spawn(move || -> Result<(), anyhow::Error> {
loop {
match rx.recv() {
std::result::Result::Ok(data) => {
// Update the root index.
flow_store.put_batch_root_list(data.root_map).unwrap();
// Update the flow database.
// This should be called before `complete_last_chunk_merkle` so that we do not save
// subtrees with data known.
flow_store.append_entries(ChunkArray {
data: data.pad_data,
start_index: data.tx_start_flow_index,
}).unwrap();
}
std::result::Result::Err(_) => {
bail!("Receiver error");
}
}
}
});
}
fn gen_proof(&self, flow_index: u64, maybe_root: Option<DataRoot>) -> Result<FlowProof> { fn gen_proof(&self, flow_index: u64, maybe_root: Option<DataRoot>) -> Result<FlowProof> {
match maybe_root { match maybe_root {
None => self.gen_proof_at_version(flow_index, None), None => self.gen_proof_at_version(flow_index, None),
@ -910,15 +964,12 @@ impl LogManager {
assert_eq!(pad_data.len(), start_index * ENTRY_SIZE); assert_eq!(pad_data.len(), start_index * ENTRY_SIZE);
} }
// Update the root index.
self.flow_store.put_batch_root_list(root_map)?;
// Update the flow database.
// This should be called before `complete_last_chunk_merkle` so that we do not save
// subtrees with data known.
let data_size = pad_data.len() / ENTRY_SIZE; let data_size = pad_data.len() / ENTRY_SIZE;
self.flow_store.append_entries(ChunkArray { self.sender.send(UpdateFlowMessage {
data: pad_data, root_map,
start_index: tx_start_flow_index, pad_data: pad_data.to_vec(),
tx_start_flow_index,
})?; })?;
tx_start_flow_index += data_size as u64; tx_start_flow_index += data_size as u64;
if let Some(index) = completed_chunk_index { if let Some(index) = completed_chunk_index {

View File

@ -76,6 +76,14 @@ pub trait LogStoreRead: LogStoreChunkRead {
/// Return flow root and length. /// Return flow root and length.
fn get_context(&self) -> Result<(DataRoot, u64)>; fn get_context(&self) -> Result<(DataRoot, u64)>;
fn pull_seal_chunk(&self, seal_index_max: usize) -> Result<Option<Vec<SealTask>>>;
fn get_num_entries(&self) -> Result<u64>;
fn load_sealed_data(&self, chunk_index: u64) -> Result<Option<MineLoadChunk>>;
fn get_shard_config(&self) -> ShardConfig;
} }
pub trait LogStoreChunkRead { pub trait LogStoreChunkRead {
@ -145,6 +153,10 @@ pub trait LogStoreWrite: LogStoreChunkWrite {
) -> Result<bool>; ) -> Result<bool>;
fn delete_block_hash_by_number(&self, block_number: u64) -> Result<()>; fn delete_block_hash_by_number(&self, block_number: u64) -> Result<()>;
fn update_shard_config(&self, shard_config: ShardConfig);
fn submit_seal_result(&self, answers: Vec<SealAnswer>) -> Result<()>;
} }
pub trait LogStoreChunkWrite { pub trait LogStoreChunkWrite {
@ -168,20 +180,15 @@ pub trait LogChunkStore: LogStoreChunkRead + LogStoreChunkWrite + Send + Sync +
impl<T: LogStoreChunkRead + LogStoreChunkWrite + Send + Sync + 'static> LogChunkStore for T {} impl<T: LogStoreChunkRead + LogStoreChunkWrite + Send + Sync + 'static> LogChunkStore for T {}
pub trait Store: pub trait Store:
LogStoreRead + LogStoreWrite + LogStoreInner + config::Configurable + Send + Sync + 'static LogStoreRead + LogStoreWrite + config::Configurable + Send + Sync + 'static
{ {
} }
impl< impl<
T: LogStoreRead + LogStoreWrite + LogStoreInner + config::Configurable + Send + Sync + 'static, T: LogStoreRead + LogStoreWrite + config::Configurable + Send + Sync + 'static,
> Store for T > Store for T
{ {
} }
pub trait LogStoreInner {
fn flow(&self) -> &dyn Flow;
fn flow_mut(&mut self) -> &mut dyn Flow;
}
pub struct MineLoadChunk { pub struct MineLoadChunk {
// Use `Vec` instead of array to avoid thread stack overflow. // Use `Vec` instead of array to avoid thread stack overflow.
pub loaded_chunk: Vec<[u8; BYTES_PER_SEAL]>, pub loaded_chunk: Vec<[u8; BYTES_PER_SEAL]>,

View File

@ -478,7 +478,7 @@ impl SerialSyncController {
metrics::SERIAL_SYNC_SEGMENT_LATENCY.update_since(since.0); metrics::SERIAL_SYNC_SEGMENT_LATENCY.update_since(since.0);
let shard_config = self.store.get_store().flow().get_shard_config(); let shard_config = self.store.get_store().get_shard_config();
let next_chunk = segment_to_sector(shard_config.next_segment_index( let next_chunk = segment_to_sector(shard_config.next_segment_index(
sector_to_segment(from_chunk), sector_to_segment(from_chunk),
sector_to_segment(self.tx_start_chunk_in_flow), sector_to_segment(self.tx_start_chunk_in_flow),

View File

@ -807,7 +807,7 @@ impl SyncService {
} }
async fn tx_sync_start_index(store: &Store, tx: &Transaction) -> Result<Option<u64>> { async fn tx_sync_start_index(store: &Store, tx: &Transaction) -> Result<Option<u64>> {
let shard_config = store.get_store().flow().get_shard_config(); let shard_config = store.get_store().get_shard_config();
let start_segment = sector_to_segment(tx.start_entry_index()); let start_segment = sector_to_segment(tx.start_entry_index());
let end = let end =
bytes_to_chunks(usize::try_from(tx.size).map_err(|e| anyhow!("tx size e={}", e))?); bytes_to_chunks(usize::try_from(tx.size).map_err(|e| anyhow!("tx size e={}", e))?);