Aggregator setup pipelining #487
Conversation
ozankaymak
requested review from
mmtftr,
ekrembal,
ceyhunsen and
ozankaymak
and removed request for
a team
There was a problem hiding this comment.
Thanks a lot for the changes.
I believe the original issue also wants the try_join_alls to happen in background tasks so that they're happening in parallel (you can check aggregator for details)
We're also going to be working on moving the business logic out of the gRPC handlers, and moving parsing/validation logic out into trait impls so that they're more easily tested and the code has less clutter.
|
Thank you for the comments @mmtftr , and sorry for my late reply! At this point I see that |
doscortados
force-pushed
the
setup-pipelines
branch
from
63439bb to
fac7127
Compare
core/src/rpc/aggregator.rs
Outdated
| watchtower_params_setup_task_handle, | ||
| ]) | ||
| .await | ||
| .map_err(|e| BridgeError::Error(format!("aggregator setup failed: {e:?}")))?; |
There was a problem hiding this comment.
Dear reviewer,
Please suggest better error type here, this one seems like a too general one.
There was a problem hiding this comment.
Maybe BridgeError::RPCStreamEndedUnexpectedly? Or we can create a new one in core/src/rpc/error.rs.
core/src/rpc/aggregator.rs
Outdated
| watchtower_params_setup_task_handle, | ||
| ]) | ||
| .await | ||
| .map_err(|e| BridgeError::Error(format!("aggregator setup failed: {e:?}")))?; |
There was a problem hiding this comment.
Maybe BridgeError::RPCStreamEndedUnexpectedly? Or we can create a new one in core/src/rpc/error.rs.
|
|
||
| future.await?; // TODO: This is dangerous: If channel size becomes not sufficient, this will block forever. | ||
| let operator_params_setup_task_handle = tokio::spawn(async move { | ||
| for response in operator_params_response_streams.iter_mut() { |
There was a problem hiding this comment.
Currently this loop:
- For each operator
- Stream all messages to operator (await)
This will block for each operator and won't pipeline operators in parallel.
can we collect all async blocks on the inside and join them on the outside of this loop?
| .map_err(|_| output_stream_ended_prematurely())?; | ||
| let watchtower_params_setup_task_handle = | ||
| tokio::spawn(async move { | ||
| for response in watchtower_params_response_streams.iter_mut() { |
There was a problem hiding this comment.
same as above, let's collect the inner blocks and join on the outside
Closes #460