From 01c2dd1135289aebdf945cfc598c804798661a52 Mon Sep 17 00:00:00 2001 From: MiniFrenchBread <103425574+MiniFrenchBread@users.noreply.github.com> Date: Tue, 6 Feb 2024 17:51:31 +0800 Subject: [PATCH] feat: upload_segments (#21) * feat: upload_segments * feat: max request body --- .../src/mem_pool/chunk_pool_inner.rs | 10 +- node/rpc/src/config.rs | 1 + node/rpc/src/lib.rs | 10 +- node/rpc/src/zgs/api.rs | 3 + node/rpc/src/zgs/impl.rs | 100 ++++++++++-------- node/src/client/builder.rs | 2 +- node/src/config/convert.rs | 1 + node/src/config/mod.rs | 1 + 8 files changed, 79 insertions(+), 49 deletions(-) diff --git a/node/chunk_pool/src/mem_pool/chunk_pool_inner.rs b/node/chunk_pool/src/mem_pool/chunk_pool_inner.rs index 6fa6da3..c224c54 100644 --- a/node/chunk_pool/src/mem_pool/chunk_pool_inner.rs +++ b/node/chunk_pool/src/mem_pool/chunk_pool_inner.rs @@ -213,6 +213,10 @@ impl MemoryChunkPool { /// Updates the cached file info when log entry retrieved from blockchain. pub async fn update_file_info(&self, tx: &Transaction) -> Result { + info!( + "start to flush cached segments to log store. data root: {}, tx_seq:{}", + tx.data_merkle_root, tx.seq + ); let maybe_file = self .inner .lock() @@ -225,7 +229,7 @@ impl MemoryChunkPool { self.write_chunks( SegmentInfo { root: tx.data_merkle_root, - seg_data: seg.data.clone(), + seg_data: seg.data, seg_proof: proof, seg_index, chunks_per_segment: file.chunks_per_segment, @@ -236,6 +240,10 @@ impl MemoryChunkPool { .await? } } + info!( + "cached segments flushed to log store. data root: {}, tx_seq:{}", + tx.data_merkle_root, tx.seq + ); Ok(true) } diff --git a/node/rpc/src/config.rs b/node/rpc/src/config.rs index 2b91ad1..902a096 100644 --- a/node/rpc/src/config.rs +++ b/node/rpc/src/config.rs @@ -6,5 +6,6 @@ pub struct Config { pub listen_address: SocketAddr, pub listen_address_admin: Option, pub chunks_per_segment: usize, + pub max_request_body_size: u32, pub max_cache_file_size: usize, } diff --git a/node/rpc/src/lib.rs b/node/rpc/src/lib.rs index 17b1710..cb4a3fd 100644 --- a/node/rpc/src/lib.rs +++ b/node/rpc/src/lib.rs @@ -74,6 +74,10 @@ pub async fn run_server( Ok(handles) } +fn server_builder(ctx: Context) -> HttpServerBuilder { + HttpServerBuilder::default().max_request_body_size(ctx.config.max_request_body_size) +} + /// Run a single RPC server for all namespace RPCs. async fn run_server_all(ctx: Context) -> Result> { // public rpc @@ -89,7 +93,7 @@ async fn run_server_all(ctx: Context) -> Result zgs.merge(mine)?; } - Ok(HttpServerBuilder::default() + Ok(server_builder(ctx.clone()) .build(ctx.config.listen_address) .await? .start(zgs)?) @@ -112,12 +116,12 @@ async fn run_server_public_private( admin.merge(mine)?; } - let handle_public = HttpServerBuilder::default() + let handle_public = server_builder(ctx.clone()) .build(ctx.config.listen_address) .await? .start(zgs)?; - let handle_private = HttpServerBuilder::default() + let handle_private = server_builder(ctx.clone()) .build(listen_addr_private) .await? .start(admin)?; diff --git a/node/rpc/src/zgs/api.rs b/node/rpc/src/zgs/api.rs index f1f6c75..5bf22e1 100644 --- a/node/rpc/src/zgs/api.rs +++ b/node/rpc/src/zgs/api.rs @@ -11,6 +11,9 @@ pub trait Rpc { #[method(name = "uploadSegment")] async fn upload_segment(&self, segment: SegmentWithProof) -> RpcResult<()>; + #[method(name = "uploadSegments")] + async fn upload_segments(&self, segments: Vec) -> RpcResult<()>; + #[method(name = "downloadSegment")] async fn download_segment( &self, diff --git a/node/rpc/src/zgs/impl.rs b/node/rpc/src/zgs/impl.rs index b7b130c..466e3d2 100644 --- a/node/rpc/src/zgs/impl.rs +++ b/node/rpc/src/zgs/impl.rs @@ -35,52 +35,14 @@ impl RpcServer for RpcServerImpl { async fn upload_segment(&self, segment: SegmentWithProof) -> RpcResult<()> { debug!(root = %segment.root, index = %segment.index, "zgs_uploadSegment"); + self.put_segment(segment).await + } - let _ = self.ctx.chunk_pool.validate_segment_size(&segment.data)?; - - let maybe_tx = self - .ctx - .log_store - .get_tx_by_data_root(&segment.root) - .await?; - let mut need_cache = false; - - if self - .ctx - .chunk_pool - .check_already_has_cache(&segment.root) - .await - { - need_cache = true; + async fn upload_segments(&self, segments: Vec) -> RpcResult<()> { + debug!("zgs_uploadSegments()"); + for segment in segments.into_iter() { + self.put_segment(segment).await?; } - - if !need_cache { - need_cache = self.check_need_cache(&maybe_tx, segment.file_size).await?; - } - - segment.validate(self.ctx.config.chunks_per_segment)?; - - let seg_info = SegmentInfo { - root: segment.root, - seg_data: segment.data, - seg_proof: segment.proof, - seg_index: segment.index, - chunks_per_segment: self.ctx.config.chunks_per_segment, - }; - - if need_cache { - self.ctx.chunk_pool.cache_chunks(seg_info).await?; - } else { - let file_id = FileID { - root: seg_info.root, - tx_id: maybe_tx.unwrap().id(), - }; - self.ctx - .chunk_pool - .write_chunks(seg_info, file_id, segment.file_size) - .await?; - } - Ok(()) } @@ -236,4 +198,54 @@ impl RpcServerImpl { uploaded_seg_num, }) } + + async fn put_segment(&self, segment: SegmentWithProof) -> RpcResult<()> { + debug!(root = %segment.root, index = %segment.index, "putSegment"); + + self.ctx.chunk_pool.validate_segment_size(&segment.data)?; + + let maybe_tx = self + .ctx + .log_store + .get_tx_by_data_root(&segment.root) + .await?; + let mut need_cache = false; + + if self + .ctx + .chunk_pool + .check_already_has_cache(&segment.root) + .await + { + need_cache = true; + } + + if !need_cache { + need_cache = self.check_need_cache(&maybe_tx, segment.file_size).await?; + } + + segment.validate(self.ctx.config.chunks_per_segment)?; + + let seg_info = SegmentInfo { + root: segment.root, + seg_data: segment.data, + seg_proof: segment.proof, + seg_index: segment.index, + chunks_per_segment: self.ctx.config.chunks_per_segment, + }; + + if need_cache { + self.ctx.chunk_pool.cache_chunks(seg_info).await?; + } else { + let file_id = FileID { + root: seg_info.root, + tx_id: maybe_tx.unwrap().id(), + }; + self.ctx + .chunk_pool + .write_chunks(seg_info, file_id, segment.file_size) + .await?; + } + Ok(()) + } } diff --git a/node/src/client/builder.rs b/node/src/client/builder.rs index 53c73de..37275fe 100644 --- a/node/src/client/builder.rs +++ b/node/src/client/builder.rs @@ -219,7 +219,7 @@ impl ClientBuilder { let async_store = require!("rpc", self, async_store).clone(); let network_send = require!("rpc", self, network).send.clone(); let mine_send = self.miner.as_ref().map(|x| x.send.clone()); - let synced_tx_recv = require!("sync", self, log_sync).send.subscribe(); + let synced_tx_recv = require!("rpc", self, log_sync).send.subscribe(); let (chunk_pool, chunk_pool_handler) = chunk_pool::unbounded(chunk_pool_config, async_store.clone(), network_send.clone()); diff --git a/node/src/config/convert.rs b/node/src/config/convert.rs index 1e0f0bb..14c9f4d 100644 --- a/node/src/config/convert.rs +++ b/node/src/config/convert.rs @@ -77,6 +77,7 @@ impl ZgsConfig { enabled: self.rpc_enabled, listen_address, listen_address_admin, + max_request_body_size: self.max_request_body_size, chunks_per_segment: self.rpc_chunks_per_segment, max_cache_file_size: self.rpc_max_cache_file_size, }) diff --git a/node/src/config/mod.rs b/node/src/config/mod.rs index e217d24..b859fa7 100644 --- a/node/src/config/mod.rs +++ b/node/src/config/mod.rs @@ -38,6 +38,7 @@ build_config! { (rpc_enabled, (bool), true) (rpc_listen_address, (String), "127.0.0.1:5678".to_string()) (rpc_listen_address_admin, (String), "127.0.0.1:5679".to_string()) + (max_request_body_size, (u32), 100*1024*1024) // 100MB (rpc_chunks_per_segment, (usize), 1024) (rpc_max_cache_file_size, (usize), 10*1024*1024) //10MB