fn download_next_compute_work(
        &self,
        peer_id: NodeId,
    ) -> Result<Vec<GossipChunkRequest>, impl Error> {
        // Get the peer context.
        let mut current_peers = self.current_peers.lock().unwrap();
        let peer_context = self.is_peer_ready_for_download(peer_id, &current_peers)?;
        let requested_instant = Instant::now(); // function granularity for instant is good enough
        let max_streams_per_peer = self.gossip_config.max_artifact_streams_per_peer as usize;

        assert!(peer_context.requested.len() <= max_streams_per_peer);
        let num_downloadable_chunks = max_streams_per_peer - peer_context.requested.len();
        if num_downloadable_chunks == 0 {
            return Err(Box::new(P2PError {
                p2p_error_code: P2PErrorCode::Busy,
            }));
        }

        let mut requests = Vec::new();
        let mut artifacts_under_construction = self.artifacts_under_construction.write().unwrap();
        // Get a prioritized iterator.
        let peer_advert_queues = self.prioritizer.get_peer_priority_queues(peer_id);
        let peer_advert_map = peer_advert_queues.peer_advert_map_ref.read().unwrap();

        let mut visited = 0;
        for (_, advert_tracker) in peer_advert_map.iter() {
            visited += 1;
            if requests.len() >= num_downloadable_chunks {
                break;
            }

            let mut advert_tracker = advert_tracker.write().unwrap();
            let advert_tracker = advert_tracker.deref_mut();

            // Try to begin a download for the artifact and collect its chunk requests.
            if let Some(artifact_tracker) = artifacts_under_construction.schedule_download(
                peer_id,
                &advert_tracker.advert,
                &self.gossip_config,
                current_peers.len() as u32,
                self.artifact_manager.as_ref(),
            ) {
                // Collect gossip requests that can be initiated for this artifact.
                // The function get_chunk_request() returns requests for chunks that satisfy
                // chunk download constraints. These requests are collected and download
                // attempts are recorded.
                let new_chunk_requests = artifact_tracker
                    .chunkable
                    .chunks_to_download()
                    .filter_map(|id: ChunkId| {
                        self.get_chunk_request(&current_peers, peer_id, advert_tracker, id)
                            .map(|req| {
                                advert_tracker.record_attempt(id, &peer_id);
                                req
                            })
                    })
                    .take(num_downloadable_chunks - requests.len());

                // Extend the requests to be send out to this peer
                requests.extend(new_chunk_requests);
            }
        }

        self.metrics.download_next_visited.set(visited as i64);
        self.metrics
            .download_next_selected
            .set(requests.len() as i64);

        let peer_context = current_peers.get_mut(&peer_id).unwrap();
        peer_context.requested.extend(requests.iter().map(|req| {
            (
                GossipRequestTrackerKey {
                    artifact_id: req.artifact_id.clone(),
                    integrity_hash: req.integrity_hash.clone(),
                    chunk_id: req.chunk_id,
                },
                GossipRequestTracker { requested_instant },
            )
        }));

        assert!(peer_context.requested.len() <= max_streams_per_peer);
        Ok(requests)
    }
/// The method returns the next set of downloads that can be initiated
/// within the constraints of the ICP protocol.
The method returns the next set of downloads that can be initiated
within the constraints of the ICP protocol.
The method returns the next set of downloads that can be initiated within the constraints of the ICP protocol .
fn download_next_compute_work(
        &self,
        peer_id: NodeId,
    ) -> Result<Vec<GossipChunkRequest>, impl Error> {
        // Get the peer context.
        let mut current_peers = self.current_peers.lock().unwrap();
        let peer_context = self.is_peer_ready_for_download(peer_id, &current_peers)?;
        let requested_instant = Instant::now(); // function granularity for instant is good enough
        let max_streams_per_peer = self.gossip_config.max_artifact_streams_per_peer as usize;
        assert!(peer_context.requested.len() <= max_streams_per_peer);
        let num_downloadable_chunks = max_streams_per_peer - peer_context.requested.len();
        if num_downloadable_chunks == 0 {
            return Err(Box::new(P2PError {
                p2p_error_code: P2PErrorCode::Busy,
            }));
        }
        let mut requests = Vec::new();
        let mut artifacts_under_construction = self.artifacts_under_construction.write().unwrap();
        // Get a prioritized iterator.
        let peer_advert_queues = self.prioritizer.get_peer_priority_queues(peer_id);
        let peer_advert_map = peer_advert_queues.peer_advert_map_ref.read().unwrap();
        let mut visited = 0;
        for (_, advert_tracker) in peer_advert_map.iter() {
            visited += 1;
            if requests.len() >= num_downloadable_chunks {
                break;
            }
            let mut advert_tracker = advert_tracker.write().unwrap();
            let advert_tracker = advert_tracker.deref_mut();
            // Try to begin a download for the artifact and collect its chunk requests.
            if let Some(artifact_tracker) = artifacts_under_construction.schedule_download(
                peer_id,
                &advert_tracker.advert,
                &self.gossip_config,
                current_peers.len() as u32,
                self.artifact_manager.as_ref(),
            ) {
                // Collect gossip requests that can be initiated for this artifact.
                // The function get_chunk_request() returns requests for chunks that satisfy
                // chunk download constraints. These requests are collected and download
                // attempts are recorded.
                let new_chunk_requests = artifact_tracker
                    .chunkable
                    .chunks_to_download()
                    .filter_map(|id: ChunkId| {
                        self.get_chunk_request(&current_peers, peer_id, advert_tracker, id)
                            .map(|req| {
                                advert_tracker.record_attempt(id, &peer_id);
                                req
                            })
                    })
                    .take(num_downloadable_chunks - requests.len());
                // Extend the requests to be send out to this peer
                requests.extend(new_chunk_requests);
            }
        }
        self.metrics.download_next_visited.set(visited as i64);
        self.metrics
            .download_next_selected
            .set(requests.len() as i64);
        let peer_context = current_peers.get_mut(&peer_id).unwrap();
        peer_context.requested.extend(requests.iter().map(|req| {
            (
                GossipRequestTrackerKey {
                    artifact_id: req.artifact_id.clone(),
                    integrity_hash: req.integrity_hash.clone(),
                    chunk_id: req.chunk_id,
                },
                GossipRequestTracker { requested_instant },
            )
        }));
        assert!(peer_context.requested.len() <= max_streams_per_peer);
        Ok(requests)
    }
fn download_next_compute_work ( & self , peer_id : NodeId , ) -> Result < Vec < GossipChunkRequest > , impl Error > { // Get the peer context. let mut current_peers = self . current_peers . lock ( ) . unwrap ( ) ; let peer_context = self . is_peer_ready_for_download ( peer_id , & current_peers ) ? ; let requested_instant = Instant :: now ( ) ; // function granularity for instant is good enough let max_streams_per_peer = self . gossip_config . max_artifact_streams_per_peer as usize ; assert ! ( peer_context requested len ( ) max_streams_per_peer ) ; let num_downloadable_chunks = max_streams_per_peer - peer_context . requested . len ( ) ; if num_downloadable_chunks == 0 { return Err ( Box :: new ( P2PError { p2p_error_code : P2PErrorCode :: Busy , } ) ) ; } let mut requests = Vec :: new ( ) ; let mut artifacts_under_construction = self . artifacts_under_construction . write ( ) . unwrap ( ) ; // Get a prioritized iterator. let peer_advert_queues = self . prioritizer . get_peer_priority_queues ( peer_id ) ; let peer_advert_map = peer_advert_queues . peer_advert_map_ref . read ( ) . unwrap ( ) ; let mut visited = 0 ; for ( _ , advert_tracker ) in peer_advert_map . iter ( ) { visited += 1 ; if requests . len ( ) >= num_downloadable_chunks { break ; } let mut advert_tracker = advert_tracker . write ( ) . unwrap ( ) ; let advert_tracker = advert_tracker . deref_mut ( ) ; // Try to begin a download for the artifact and collect its chunk requests. if let Some ( artifact_tracker ) = artifacts_under_construction . schedule_download ( peer_id , & advert_tracker . advert , & self . gossip_config , current_peers . len ( ) as u32 , self . artifact_manager . as_ref ( ) , ) { // Collect gossip requests that can be initiated for this artifact. // The function get_chunk_request() returns requests for chunks that satisfy // chunk download constraints. These requests are collected and download // attempts are recorded. let new_chunk_requests = artifact_tracker . chunkable . chunks_to_download ( ) . filter_map ( | id : ChunkId | { self . get_chunk_request ( & current_peers , peer_id , advert_tracker , id ) . map ( | req | { advert_tracker . record_attempt ( id , & peer_id ) ; req } ) } ) . take ( num_downloadable_chunks - requests . len ( ) ) ; // Extend the requests to be send out to this peer requests . extend ( new_chunk_requests ) ; } } self . metrics . download_next_visited . set ( visited as i64 ) ; self . metrics . download_next_selected . set ( requests . len ( ) as i64 ) ; let peer_context = current_peers . get_mut ( & peer_id ) . unwrap ( ) ; peer_context . requested . extend ( requests . iter ( ) . map ( | req | { ( GossipRequestTrackerKey { artifact_id : req . artifact_id . clone ( ) , integrity_hash : req . integrity_hash . clone ( ) , chunk_id : req . chunk_id , } , GossipRequestTracker { requested_instant } , ) } ) ) ; assert ! ( peer_context requested len ( ) max_streams_per_peer ) ; Ok ( requests ) }
The method returns the next set of downloads that can be initiated
within the constraints of the ICP protocol.
The method returns the next set of downloads that can be initiated within the constraints of the ICP protocol .
&self
peer_id
NodeId
&self
peer_id
NodeId