/// Queries all subnets for new slices and puts / appends them to the pool.
Queries all subnets for new slices and puts / appends them to the pool.
Queries all subnets for new slices and puts / appends them to the pool .
async fn refill_pool(&self, pool_byte_size_soft_cap: usize, slice_byte_size_max: usize) {
        let pool_slice_stats = {
            let pool = self.pool.lock().unwrap();
            if pool.byte_size() > pool_byte_size_soft_cap {
                // Abort if pool is already full.
                return;
            }
            pool.peers()
                // Skip our own subnet, the loopback stream is routed separately.
                .filter(|&&subnet_id| subnet_id != self.endpoint_resolver.subnet_id)
                .map(|&subnet_id| (subnet_id, pool.slice_stats(subnet_id)))
                .collect::<BTreeMap<_, _>>()
        };
        for (subnet_id, slice_stats) in pool_slice_stats {
            let (stream_position, messages_begin, msg_count, byte_size) = match slice_stats {
                // Have a cached stream position.
                (Some(stream_position), messages_begin, msg_count, byte_size) => {
                    (stream_position, messages_begin, msg_count, byte_size)
                }
                // No cached stream position, no pooling / refill necessary.
                (None, _, _, _) => continue,
            };
            let (witness_begin, msg_begin, slice_byte_limit) = match messages_begin {
                // Existing pooled stream, pull partial slice and append.
                Some(messages_begin) if messages_begin == stream_position.message_index => (
                    stream_position.message_index,
                    stream_position.message_index + (msg_count as u64).into(),
                    slice_byte_size_max.saturating_sub(byte_size),
                ),
                // No pooled stream, or pooled stream does not begin at cached stream position, pull
                // complete slice from cached stream position.
                _ => (
                    stream_position.message_index,
                    stream_position.message_index,
                    slice_byte_size_max,
                ),
            };
            if slice_byte_limit < SLICE_BYTE_SIZE_MIN {
                // No more space left in the pool for this slice, bail out.
                continue;
            }
            // `XNetEndpoint` URL of a node on `subnet_id`.
            let endpoint_locator = match self.endpoint_resolver.xnet_endpoint_url(
                subnet_id,
                witness_begin,
                msg_begin,
                // XNetEndpoint only counts message bytes, allow some overhead (measuread: 350
                // bytes for certification plus base witness, 2% for large payloads).
                (slice_byte_limit.saturating_sub(350)) * 98 / 100,
            ) {
                Ok(endpoint_locator) => endpoint_locator,
                Err(e) => {
                    log!(self.log, e.log_level(), "{}", e);
                    self.metrics.observe_pull_attempt(e.to_label_value());
                    continue;
                }
            };
            // Spawn an async task to query the `XNetEndpoint` on `subnet_id`.
            let xnet_client = self.xnet_client.clone();
            let metrics = Arc::clone(&self.metrics);
            let pool = Arc::clone(&self.pool);
            let log = self.log.clone();
            self.runtime_handle.spawn(async move {
                let timer = Timer::start();
                metrics.outstanding_queries.inc();
                let query_result = xnet_client.query(&endpoint_locator).await;
                metrics.outstanding_queries.dec();
                let proximity = endpoint_locator.proximity.into();
                match query_result {
                    Ok(slice) => {
                        let res = if witness_begin != msg_begin {
                            // Pulled a stream suffix, append to pooled slice.
                            pool.lock().unwrap().append(subnet_id, slice)
                        } else {
                            // Pulled a complete stream, replace polled slice (if any).
                            pool.lock().unwrap().put(subnet_id, slice)
                        };
                        let status = match res {
                            Ok(()) => STATUS_SUCCESS,
                            Err(e) => e.to_label_value(),
                        };
                        metrics.observe_query_slice_duration(status, proximity, timer);
                        metrics.observe_pull_attempt(status);
                    }
                    Err(e) => {
                        metrics.observe_query_slice_duration(&e.to_label_value(), proximity, timer);
                        metrics.observe_pull_attempt(&e.to_label_value());
                        if let XNetClientError::NoContent = e {
                        } else if Self::pass_log_sampling() {
                            info!(
                                log,
                                "Failed to query stream slice for subnet {} from node {}: {}",
                                subnet_id,
                                endpoint_locator.node_id,
                                e
                            );
                        }
                    }
                }
            });
        }
    }
async fn refill_pool ( & self , pool_byte_size_soft_cap : usize , slice_byte_size_max : usize ) { let pool_slice_stats = { let pool = self . pool . lock ( ) . unwrap ( ) ; if pool . byte_size ( ) > pool_byte_size_soft_cap { // Abort if pool is already full. return ; } pool . peers ( ) // Skip our own subnet, the loopback stream is routed separately. . filter ( | & & subnet_id | subnet_id != self . endpoint_resolver . subnet_id ) . map ( | & subnet_id | ( subnet_id , pool . slice_stats ( subnet_id ) ) ) . collect :: < BTreeMap < _ , _ > > ( ) } ; for ( subnet_id , slice_stats ) in pool_slice_stats { let ( stream_position , messages_begin , msg_count , byte_size ) = match slice_stats { // Have a cached stream position. ( Some ( stream_position ) , messages_begin , msg_count , byte_size ) => { ( stream_position , messages_begin , msg_count , byte_size ) } // No cached stream position, no pooling / refill necessary. ( None , _ , _ , _ ) => continue , } ; let ( witness_begin , msg_begin , slice_byte_limit ) = match messages_begin { // Existing pooled stream, pull partial slice and append. Some ( messages_begin ) if messages_begin == stream_position . message_index => ( stream_position . message_index , stream_position . message_index + ( msg_count as u64 ) . into ( ) , slice_byte_size_max . saturating_sub ( byte_size ) , ) , // No pooled stream, or pooled stream does not begin at cached stream position, pull // complete slice from cached stream position. _ => ( stream_position . message_index , stream_position . message_index , slice_byte_size_max , ) , } ; if slice_byte_limit < SLICE_BYTE_SIZE_MIN { // No more space left in the pool for this slice, bail out. continue ; } // `XNetEndpoint` URL of a node on `subnet_id`. let endpoint_locator = match self . endpoint_resolver . xnet_endpoint_url ( subnet_id , witness_begin , msg_begin , // XNetEndpoint only counts message bytes, allow some overhead (measuread: 350 // bytes for certification plus base witness, 2% for large payloads). ( slice_byte_limit . saturating_sub ( 350 ) ) * 98 / 100 , ) { Ok ( endpoint_locator ) => endpoint_locator , Err ( e ) => { log ! ( self log e log_level ( ) " " e ) ; self . metrics . observe_pull_attempt ( e . to_label_value ( ) ) ; continue ; } } ; // Spawn an async task to query the `XNetEndpoint` on `subnet_id`. let xnet_client = self . xnet_client . clone ( ) ; let metrics = Arc :: clone ( & self . metrics ) ; let pool = Arc :: clone ( & self . pool ) ; let log = self . log . clone ( ) ; self . runtime_handle . spawn ( async move { let timer = Timer :: start ( ) ; metrics . outstanding_queries . inc ( ) ; let query_result = xnet_client . query ( & endpoint_locator ) . await ; metrics . outstanding_queries . dec ( ) ; let proximity = endpoint_locator . proximity . into ( ) ; match query_result { Ok ( slice ) => { let res = if witness_begin != msg_begin { // Pulled a stream suffix, append to pooled slice. pool . lock ( ) . unwrap ( ) . append ( subnet_id , slice ) } else { // Pulled a complete stream, replace polled slice (if any). pool . lock ( ) . unwrap ( ) . put ( subnet_id , slice ) } ; let status = match res { Ok ( ( ) ) => STATUS_SUCCESS , Err ( e ) => e . to_label_value ( ) , } ; metrics . observe_query_slice_duration ( status , proximity , timer ) ; metrics . observe_pull_attempt ( status ) ; } Err ( e ) => { metrics . observe_query_slice_duration ( & e . to_label_value ( ) , proximity , timer ) ; metrics . observe_pull_attempt ( & e . to_label_value ( ) ) ; if let XNetClientError :: NoContent = e { } else if Self :: pass_log_sampling ( ) { info ! ( log " " subnet_id endpoint_locator node_id e ) ; } } } } ) ; } }
Queries all subnets for new slices and puts / appends them to the pool.
Queries all subnets for new slices and puts / appends them to the pool .
&self
pool_byte_size_soft_cap
usize
slice_byte_size_max
usize