Skip to content
This repository was archived by the owner on Nov 15, 2023. It is now read-only.

Fill up request slots via launch_parallel_requests #3681

Merged
eskimor merged 2 commits into from
Aug 24, 2021
Merged

Fill up request slots via launch_parallel_requests #3681

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8dd4c1e
Fill up requests slots via `launch_parallel_requests`
eskimor Aug 20, 2021
90f5c4d
Merge branch 'master' into rk-more-parallel-availability-recovery
eskimor Aug 23, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions node/network/availability-recovery/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
use std::{
collections::{HashMap, VecDeque},
pin::Pin,
time::Duration,
};

use futures::{
Expand All @@ -43,7 +44,7 @@ use polkadot_node_network_protocol::{
IfDisconnected, UnifiedReputationChange as Rep,
};
use polkadot_node_primitives::{AvailableData, ErasureChunk};
use polkadot_node_subsystem_util::request_session_info;
use polkadot_node_subsystem_util::{request_session_info, TimeoutExt};
use polkadot_primitives::v1::{
AuthorityDiscoveryId, BlakeTwo256, BlockNumber, CandidateHash, CandidateReceipt, GroupIndex,
Hash, HashT, SessionIndex, SessionInfo, ValidatorId, ValidatorIndex,
Expand Down Expand Up @@ -72,6 +73,10 @@ const LRU_SIZE: usize = 16;

const COST_INVALID_REQUEST: Rep = Rep::CostMajor("Peer sent unparsable request");

/// Max time we want to wait for responses, before calling `launch_parallel_requests` again to fill
/// up slots.
const MAX_CHUNK_WAIT: Duration = Duration::from_secs(1);

/// The Availability Recovery Subsystem.
pub struct AvailabilityRecoverySubsystem {
fast_path: bool,
Expand Down Expand Up @@ -285,7 +290,11 @@ impl RequestChunksPhase {

async fn wait_for_chunks(&mut self, params: &InteractionParams) {
// Wait for all current requests to conclude or time-out, or until we reach enough chunks.
while let Some(request_result) = self.requesting_chunks.next().await {
// We will also stop, if there has not been a response for `MAX_CHUNK_WAIT`, so
// `launch_parallel_requests` cann fill up slots again.
while let Some(request_result) =
self.requesting_chunks.next().timeout(MAX_CHUNK_WAIT).await.flatten()
Copy link
Contributor

@rphmeier rphmeier Aug 23, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Previously, a timeout would lead to an invocation of self.shuffling.push_front(validator_index). Now, it no longer does. This could also lead to mistaken "unavailable" returns from the process.

Copy link
Member Author

@eskimor eskimor Aug 24, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It still does. I did not change that code. The above timeout is a soft timeout, we still wait for the futures to return and they will still return with a timeout, triggering the push_front call. The only difference is that in the meantime we filled up free slots with new futures.

Copy link
Contributor

@rphmeier rphmeier Aug 24, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, I see. The timeout causes the loop to break so we do another loop iteration where we populate chunks.

{
match request_result {
Ok(Some(chunk)) => {
// Check merkle proofs of any received chunks.
Expand Down