[Bug fix]Use global sync dag store #4347
Conversation
WalkthroughThe pull request introduces changes across multiple files to update the handling of the Changes
Possibly related PRs
Suggested reviewers
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Outside diff range and nitpick comments (1)
sync/src/tasks/tests.rs (1)
920-920: Consider consistent error handling patternFor consistency with line 57, consider using the
?operator instead ofexpect.-Arc::new(SyncDagStore::create_for_testing().expect("failed to create the sync dag store")), +Arc::new(SyncDagStore::create_for_testing()?),
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (8)
sync/src/parallel/sender.rs(2 hunks)sync/src/sync.rs(4 hunks)sync/src/tasks/block_sync_task.rs(2 hunks)sync/src/tasks/continue_execute_absent_block.rs(1 hunks)sync/src/tasks/inner_sync_task.rs(2 hunks)sync/src/tasks/mock.rs(6 hunks)sync/src/tasks/mod.rs(1 hunks)sync/src/tasks/tests.rs(2 hunks)
🔇 Additional comments (12)
sync/src/tasks/block_sync_task.rs (1)
217-217: Update sync_dag_store to Arc<SyncDagStore> in BlockCollector
Changing sync_dag_store to Arc<SyncDagStore> allows for shared ownership and thread-safe access, which is appropriate for concurrent environments.
Also applies to: 267-267
sync/src/tasks/inner_sync_task.rs (1)
41-41: Update sync_dag_store to Arc<SyncDagStore> in InnerSyncTask
Using Arc<SyncDagStore> in both the struct and constructor facilitates thread-safe shared ownership, aligning with concurrent usage patterns.
Also applies to: 61-61
sync/src/tasks/continue_execute_absent_block.rs (1)
22-22: Update sync_dag_store to Arc<SyncDagStore> in ContinueExecuteAbsentBlock
Transitioning sync_dag_store to Arc<SyncDagStore> enables shared ownership and thread-safe access, which is beneficial for concurrent execution.
Also applies to: 29-29
sync/src/parallel/sender.rs (1)
29-29: Update sync_dag_store to Arc<SyncDagStore> in DagBlockSender
Modifying sync_dag_store to use Arc<SyncDagStore> facilitates thread-safe shared ownership, enhancing concurrency management within DagBlockSender.
Also applies to: 41-41
sync/src/tasks/mock.rs (3)
138-138: LGTM! Thread-safe field declaration.
The change from SyncDagStore to Arc<SyncDagStore> enhances thread safety by enabling shared ownership of the store across multiple threads.
158-161: LGTM! Proper initialization of Arc.
The initialization properly wraps the SyncDagStore in an Arc and includes appropriate error handling.
191-191: LGTM! Consistent Arc usage across constructors.
The new_with_storage and new_with_strategy methods consistently initialize the sync_dag_store as Arc<SyncDagStore>.
Also applies to: 211-211
sync/src/sync.rs (3)
65-65: LGTM! Thread-safe service field.
The addition of sync_dag_store as Arc<SyncDagStore> to the service struct enables proper shared ownership.
87-93: LGTM! Proper initialization with configuration.
The initialization properly uses configuration parameters for the sync directory and cache size, ensuring flexibility and configurability.
238-238: LGTM! Proper cloning of Arc in check_and_start_sync.
The sync_dag_store is properly cloned before being passed to the sync task, maintaining the shared ownership semantics.
sync/src/tasks/mod.rs (1)
625-625: LGTM! Thread-safe parameter type.
The change to accept Arc<SyncDagStore> aligns with the broader effort to ensure thread-safe access to the sync DAG store across the codebase.
sync/src/tasks/tests.rs (1)
57-57: LGTM!
The change correctly wraps SyncDagStore in an Arc for thread-safe reference counting, which aligns with the broader changes to enhance thread safety.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Outside diff range and nitpick comments (2)
sync/src/parallel/executor.rs (2)
17-17: UseDurationfor time constants to improve readabilityConsider defining
MAX_TOTAL_WAITING_TIMEusingDurationto enhance readability and prevent unit confusion.Apply this diff:
-use std::sync::Arc; +use std::sync::Arc; +use tokio::time::Duration; const MAX_TOTAL_WAITING_TIME: u64 = 3600000; // an hour +const MAX_TOTAL_WAITING_TIME: Duration = Duration::from_secs(3600); // an hour
101-102: UseDurationfor time variables and calculationsFor consistency and clarity, consider using
Durationfortotal_waiting_timeandwaiting_per_time. This approach improves readability and leveragesDuration's built-in methods.Apply this diff:
let handle = tokio::spawn(async move { let mut chain = None; loop { match self.receiver.recv().await { Some(op_block) => { let block = match op_block { Some(block) => block, None => { info!("sync worker channel closed"); drop(self.sender); return; } }; let header = block.header().clone(); info!( "sync parallel worker {:p} received block: {:?}", &self, block.header().id() ); - let mut total_waiting_time: u64 = 0; - let waiting_per_time: u64 = 100; + let mut total_waiting_time = Duration::ZERO; + let waiting_per_time = Duration::from_millis(100); loop { match Self::waiting_for_parents( &self.dag, self.storage.clone(), block.header().parents_hash(), ) { Ok(true) => break, Ok(false) => { if total_waiting_time >= MAX_TOTAL_WAITING_TIME { error!( "failed to check parents: {:?}, for reason: timeout", header ); match self .sender .send(ExecuteState::Error(Box::new(header.clone()))) .await { Ok(_) => (), Err(e) => { error!( "failed to send error state: {:?}, for reason: {:?}", header, e ); return; } } return; } tokio::task::yield_now().await; tokio::time::sleep(waiting_per_time).await; - total_waiting_time = - total_waiting_time.saturating_add(waiting_per_time); + total_waiting_time = total_waiting_time.checked_add(waiting_per_time).unwrap_or(MAX_TOTAL_WAITING_TIME); } Err(e) => { error!( "failed to check parents: {:?}, for reason: {:?}", header, e ); match self .sender .send(ExecuteState::Error(Box::new(header.clone()))) .await { Ok(_) => (), Err(e) => { error!( "failed to send error state: {:?}, for reason: {:?}", header, e ); return; } } return; } } }Also applies to: 133-138
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
sync/src/parallel/executor.rs(3 hunks)
🔇 Additional comments (1)
sync/src/parallel/executor.rs (1)
111-131: Reconsider the timeout duration for waiting on parent blocks
The MAX_TOTAL_WAITING_TIME is set to one hour. Depending on the application's requirements and typical network conditions, this duration might be too long, potentially delaying error handling and resource cleanup.
Please confirm that a one-hour timeout is appropriate for your use case. If not, consider adjusting it to a shorter duration that aligns with expected parent block availability.
Pull request type
Please check the type of change your PR introduces:
What is the current behavior?
Issue Number: N/A
What is the new behavior?
Other information
Summary by CodeRabbit
Release Notes
New Features
Arc<SyncDagStore>.Bug Fixes
Documentation