General code review #1
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Boilerplate version supporting only logs and transactions
Make /status report all chunks scheduled for download instead of only those currently being downloaded
kalabukdima
requested changes
Feb 13, 2024
src/main.rs
Outdated
Comment on lines
40
to
84
| async fn ping_forever( | ||
| state_manager: Arc<StateManager>, | ||
| transport: Arc<impl Transport>, | ||
| interval: Duration, | ||
| ) { | ||
| loop { | ||
| let status = state_manager.current_status().await; | ||
| let result = transport | ||
| .send_ping(transport::State { | ||
| datasets: status.available, | ||
| }) | ||
| .await; | ||
| if let Err(err) = result { | ||
| warn!("Couldn't send ping: {:?}", err); | ||
| } | ||
| tokio::time::sleep(interval).await; | ||
| } | ||
| } | ||
|
|
||
| async fn handle_updates_forever( | ||
| state_manager: Arc<StateManager>, | ||
| mut updates: impl Stream<Item = Ranges> + Unpin, | ||
| ) { | ||
| while let Some(ranges) = updates.next().await { | ||
| let result = state_manager.set_desired_ranges(ranges).await; | ||
| if let Err(err) = result { | ||
| warn!("Couldn't schedule update: {:?}", err) | ||
| } | ||
| } | ||
| unreachable!("Updates receiver closed unexpectedly"); | ||
| } | ||
|
|
||
| #[instrument(skip(state_manager, transport))] | ||
| async fn process_assignments( | ||
| state_manager: Arc<StateManager>, | ||
| transport: Arc<impl Transport>, | ||
| interval: Duration, | ||
| ) { | ||
| let state_updates = transport.stream_assignments(); | ||
| tokio::pin!(state_updates); | ||
| futures::join!( | ||
| ping_forever(state_manager.clone(), transport, interval), | ||
| handle_updates_forever(state_manager, state_updates), | ||
| ); | ||
| } |
There was a problem hiding this comment.
Should probably be moved to a separate file controller.rs.
I'm not sure what would be a good way to handle dependencies here. I don't want the Transport to know anything about the StateManager, but I want requests for status being made by a direct function call instead of one-shot channels or callbacks. One option would be to have a Controller struct that forwards status() calls to its StateManager and passes reference to itself to the Transport. That way the Transport and the StateManager only know about the Controller but not about each other.
- Move components interaction into controller.rs - Stop the process if any background task stops - Move query related code out of http_server.rs
- Don't limit concurrent downloads for individual files - Cancel dir download if any file failed to download - Cancel running downloads on shutdown
Reorganizing crate into a library was required
DataFusion is not optimizing aliased unions so things like `df.union(df.filter(...))` result in two scans. However it does optimize expressions like `df.filter(or(true, ...))` into just `df`
- Handle all swarm events in transport metrics - Remove default values for contract addresses
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
https://drive.google.com/file/d/1Q4J7wX_OT0Xrhalfwy2Tuco0TNbrqnAP/view?usp=sharing