fn dispatch(&mut self, peers: &mut HashMap<PeerId, PeerSync<B>>, protocol: &mut Context<B>) {
		if self.pending_requests.is_empty() {
			return;
		}

		let initial_pending_requests = self.pending_requests.len();

		// clean up previous failed requests so we can retry again
		for (_, requests) in self.previous_requests.iter_mut() {
			requests.retain(|(_, instant)| instant.elapsed() < JUSTIFICATION_RETRY_WAIT);
		}

		let mut available_peers = peers.iter().filter_map(|(peer, sync)| {
			// don't request to any peers that already have pending requests or are unavailable
			if sync.state != PeerSyncState::Available || self.peer_requests.contains_key(&peer) {
				None
			} else {
				Some((peer.clone(), sync.best_number))
			}
		}).collect::<VecDeque<_>>();

		let mut last_peer = available_peers.back().map(|p| p.0.clone());
		let mut unhandled_requests = VecDeque::new();

		loop {
			let (peer, peer_best_number) = match available_peers.pop_front() {
				Some(p) => p,
				_ => break,
			};

			// only ask peers that have synced past the block number that we're
			// asking the justification for and to whom we haven't already made
			// the same request recently
			let peer_eligible = {
				let request = match self.pending_requests.front() {
					Some(r) => r.clone(),
					_ => break,
				};

				peer_best_number >= request.1 &&
					!self.previous_requests
						.get(&request)
						.map(|requests| requests.iter().any(|i| i.0 == peer))
						.unwrap_or(false)
			};

			if !peer_eligible {
				available_peers.push_back((peer.clone(), peer_best_number));

				// we tried all peers and none can answer this request
				if Some(peer) == last_peer {
					last_peer = available_peers.back().map(|p| p.0.clone());

					let request = self.pending_requests.pop_front()
						.expect("verified to be Some in the beginning of the loop; qed");

					unhandled_requests.push_back(request);
				}

				continue;
			}

			last_peer = available_peers.back().map(|p| p.0.clone());

			let request = self.pending_requests.pop_front()
				.expect("verified to be Some in the beginning of the loop; qed");

			self.peer_requests.insert(peer.clone(), request);

			peers.get_mut(&peer)
				.expect("peer was is taken from available_peers; available_peers is a subset of peers; qed")
				.state = PeerSyncState::DownloadingJustification(request.0);

			trace!(target: "sync", "Requesting justification for block #{} from {}", request.0, peer);
			let request = message::generic::BlockRequest {
				id: 0,
				fields: message::BlockAttributes::JUSTIFICATION,
				from: message::FromBlock::Hash(request.0),
				to: None,
				direction: message::Direction::Ascending,
				max: Some(1),
			};

			protocol.send_block_request(peer, request);
		}

		self.pending_requests.append(&mut unhandled_requests);

		trace!(target: "sync", "Dispatched {} justification requests ({} pending)",
			initial_pending_requests - self.pending_requests.len(),
			self.pending_requests.len(),
		);
	}
/// Dispatches all possible pending requests to the given peers. Peers are
/// filtered according to the current known best block (i.e. we won't send a
/// justification request for block #10 to a peer at block #2), and we also
/// throttle requests to the same peer if a previous justification request
/// yielded no results.
Dispatches all possible pending requests to the given peers. Peers are
filtered according to the current known best block , and we also
throttle requests to the same peer if a previous justification request
yielded no results.
Dispatches all possible pending requests to the given peers . Peers are filtered according to the current known best block and we also throttle requests to the same peer if a previous justification request yielded no results .
fn dispatch(&mut self, peers: &mut HashMap<PeerId, PeerSync<B>>, protocol: &mut Context<B>) {
		if self.pending_requests.is_empty() {
			return;
		}
		let initial_pending_requests = self.pending_requests.len();
		// clean up previous failed requests so we can retry again
		for (_, requests) in self.previous_requests.iter_mut() {
			requests.retain(|(_, instant)| instant.elapsed() < JUSTIFICATION_RETRY_WAIT);
		}
		let mut available_peers = peers.iter().filter_map(|(peer, sync)| {
			// don't request to any peers that already have pending requests or are unavailable
			if sync.state != PeerSyncState::Available || self.peer_requests.contains_key(&peer) {
				None
			} else {
				Some((peer.clone(), sync.best_number))
			}
		}).collect::<VecDeque<_>>();
		let mut last_peer = available_peers.back().map(|p| p.0.clone());
		let mut unhandled_requests = VecDeque::new();
		loop {
			let (peer, peer_best_number) = match available_peers.pop_front() {
				Some(p) => p,
				_ => break,
			};
			// only ask peers that have synced past the block number that we're
			// asking the justification for and to whom we haven't already made
			// the same request recently
			let peer_eligible = {
				let request = match self.pending_requests.front() {
					Some(r) => r.clone(),
					_ => break,
				};
				peer_best_number >= request.1 &&
					!self.previous_requests
						.get(&request)
						.map(|requests| requests.iter().any(|i| i.0 == peer))
						.unwrap_or(false)
			};
			if !peer_eligible {
				available_peers.push_back((peer.clone(), peer_best_number));
				// we tried all peers and none can answer this request
				if Some(peer) == last_peer {
					last_peer = available_peers.back().map(|p| p.0.clone());
					let request = self.pending_requests.pop_front()
						.expect("verified to be Some in the beginning of the loop; qed");
					unhandled_requests.push_back(request);
				}
				continue;
			}
			last_peer = available_peers.back().map(|p| p.0.clone());
			let request = self.pending_requests.pop_front()
				.expect("verified to be Some in the beginning of the loop; qed");
			self.peer_requests.insert(peer.clone(), request);
			peers.get_mut(&peer)
				.expect("peer was is taken from available_peers; available_peers is a subset of peers; qed")
				.state = PeerSyncState::DownloadingJustification(request.0);
			trace!(target: "sync", "Requesting justification for block #{} from {}", request.0, peer);
			let request = message::generic::BlockRequest {
				id: 0,
				fields: message::BlockAttributes::JUSTIFICATION,
				from: message::FromBlock::Hash(request.0),
				to: None,
				direction: message::Direction::Ascending,
				max: Some(1),
			};
			protocol.send_block_request(peer, request);
		}
		self.pending_requests.append(&mut unhandled_requests);
		trace!(target: "sync", "Dispatched {} justification requests ({} pending)",
			initial_pending_requests - self.pending_requests.len(),
			self.pending_requests.len(),
		);
	}
fn dispatch ( & mut self , peers : & mut HashMap < PeerId , PeerSync < B > > , protocol : & mut Context < B > ) { if self . pending_requests . is_empty ( ) { return ; } let initial_pending_requests = self . pending_requests . len ( ) ; // clean up previous failed requests so we can retry again for ( _ , requests ) in self . previous_requests . iter_mut ( ) { requests . retain ( | ( _ , instant ) | instant . elapsed ( ) < JUSTIFICATION_RETRY_WAIT ) ; } let mut available_peers = peers . iter ( ) . filter_map ( | ( peer , sync ) | { // don't request to any peers that already have pending requests or are unavailable if sync . state != PeerSyncState :: Available || self . peer_requests . contains_key ( & peer ) { None } else { Some ( ( peer . clone ( ) , sync . best_number ) ) } } ) . collect :: < VecDeque < _ > > ( ) ; let mut last_peer = available_peers . back ( ) . map ( | p | p . 0 . clone ( ) ) ; let mut unhandled_requests = VecDeque :: new ( ) ; loop { let ( peer , peer_best_number ) = match available_peers . pop_front ( ) { Some ( p ) => p , _ => break , } ; // only ask peers that have synced past the block number that we're // asking the justification for and to whom we haven't already made // the same request recently let peer_eligible = { let request = match self . pending_requests . front ( ) { Some ( r ) => r . clone ( ) , _ => break , } ; peer_best_number >= request . 1 && ! self . previous_requests . get ( & request ) . map ( | requests | requests . iter ( ) . any ( | i | i . 0 == peer ) ) . unwrap_or ( false ) } ; if ! peer_eligible { available_peers . push_back ( ( peer . clone ( ) , peer_best_number ) ) ; // we tried all peers and none can answer this request if Some ( peer ) == last_peer { last_peer = available_peers . back ( ) . map ( | p | p . 0 . clone ( ) ) ; let request = self . pending_requests . pop_front ( ) . expect ( " " ) ; unhandled_requests . push_back ( request ) ; } continue ; } last_peer = available_peers . back ( ) . map ( | p | p . 0 . clone ( ) ) ; let request = self . pending_requests . pop_front ( ) . expect ( " " ) ; self . peer_requests . insert ( peer . clone ( ) , request ) ; peers . get_mut ( & peer ) . expect ( " " ) . state = PeerSyncState :: DownloadingJustification ( request . 0 ) ; trace ! ( target " " " " request 0 peer ) ; let request = message :: generic :: BlockRequest { id : 0 , fields : message :: BlockAttributes :: JUSTIFICATION , from : message :: FromBlock :: Hash ( request . 0 ) , to : None , direction : message :: Direction :: Ascending , max : Some ( 1 ) , } ; protocol . send_block_request ( peer , request ) ; } self . pending_requests . append ( & mut unhandled_requests ) ; trace ! ( target " " " " initial_pending_requests self pending_requests len ( ) self pending_requests len ( ) ) ; }
Dispatches all possible pending requests to the given peers.
Dispatches all possible pending requests to the given peers .
&mut self
peers
&mut HashMap<PeerId, PeerSync<B>>
protocol
&mut Context<B>
&mut self
peers
&mut HashMap<PeerId, PeerSync<B>>
protocol
&mut Context<B>