pub(crate) fn pop_vertex_to_add(
        &mut self,
        highway: &Highway<C>,
        pending_values: &HashMap<ProposedBlock<C>, HashSet<(ValidVertex<C>, I)>>,
        max_requests_for_vertex: usize,
    ) -> (Option<PendingVertex<I, C>>, ProtocolOutcomes<I, C>) {
        let mut outcomes = Vec::new();
        // Get the next vertex to be added; skip the ones that are already in the protocol state,
        // and the ones that are still missing dependencies.
        loop {
            let pv = match self.vertices_no_deps.pop() {
                None => return (None, outcomes),
                Some(pv) if highway.has_vertex(pv.vertex()) => continue,
                Some(pv) => pv,
            };
            if let Some(dep) = highway.missing_dependency(pv.pvv()) {
                let sender = pv.sender().clone();
                let time_received = pv.time_received;
                // Find the first dependency that `pv` needs that we haven't synchronized yet
                // and request it from the sender of `pv`. Since it relies on it, it should have
                // it as well.
                let transitive_dependency =
                    self.find_transitive_dependency(dep.clone(), &sender, time_received);
                if self
                    .vertices_no_deps
                    .contains_dependency(&transitive_dependency)
                {
                    // `dep` is already downloaded and waiting in the synchronizer queue to be
                    // added, we don't have to request it again. Add the `pv`
                    // back to the queue so that it can be retried later. `dep` does not wait for
                    // any of the dependencies currently so it should be retried soon.
                    self.add_missing_dependency(dep.clone(), pv);
                    continue;
                }
                // We are still missing a dependency. Store the vertex in the map and request
                // the dependency from the sender.
                // Make `pv` depend on the direct dependency `dep` and not `transitive_dependency`
                // since there's a higher chance of adding `pv` to the protocol
                // state after `dep` is added, rather than `transitive_dependency`.
                self.add_missing_dependency(dep.clone(), pv);
                // If we already have the dependency and it is a proposal that is currently being
                // handled by the block validator, and this sender is already known as a source,
                // do nothing.
                if pending_values
                    .values()
                    .flatten()
                    .any(|(vv, s)| vv.inner().id() == transitive_dependency && s == &sender)
                {
                    continue;
                }
                // If we already have the dependency and it is a proposal that is currently being
                // handled by the block validator, and this sender is not yet known as a source,
                // we return the proposal as if this sender had sent it to us, so they get added.
                if let Some((vv, _)) = pending_values
                    .values()
                    .flatten()
                    .find(|(vv, _)| vv.inner().id() == transitive_dependency)
                {
                    debug!(
                        dependency = ?transitive_dependency, %sender,
                        "adding sender as a source for proposal"
                    );
                    let dep_pv = PendingVertex::new(sender, vv.clone().into(), time_received);
                    // We found the next vertex to add.
                    if !self.vertices_no_deps.is_empty() {
                        // There are still vertices in the queue: schedule next call.
                        outcomes.push(ProtocolOutcome::QueueAction(ACTION_ID_VERTEX));
                    }
                    return (Some(dep_pv), outcomes);
                }
                // If we have already requested the dependency from this peer, or from the maximum
                // number of peers, do nothing.
                let entry = self
                    .requests_sent
                    .entry(transitive_dependency.clone())
                    .or_default();
                if entry.len() >= max_requests_for_vertex || !entry.insert(sender.clone()) {
                    continue;
                }
                // Otherwise request the missing dependency from the sender.
                let uuid = thread_rng().next_u64();
                debug!(?uuid, dependency = ?transitive_dependency, %sender, "requesting dependency");
                let ser_msg =
                    HighwayMessage::RequestDependency(uuid, transitive_dependency).serialize();
                outcomes.push(ProtocolOutcome::CreatedTargetedMessage(ser_msg, sender));
                continue;
            }
            // We found the next vertex to add.
            if !self.vertices_no_deps.is_empty() {
                // There are still vertices in the queue: schedule next call.
                outcomes.push(ProtocolOutcome::QueueAction(ACTION_ID_VERTEX));
            }
            return (Some(pv), outcomes);
        }
    }
/// Pops and returns the next entry from `vertices_to_be_added` that is not yet in the protocol
/// state. Also returns a `ProtocolOutcome` that schedules the next action to add a vertex,
/// unless the queue is empty, and `ProtocolOutcome`s to request missing dependencies.
Pops and returns the next entry from `vertices_to_be_added` that is not yet in the protocol
state. Also returns a `ProtocolOutcome` that schedules the next action to add a vertex,
unless the queue is empty, and `ProtocolOutcome`s to request missing dependencies.
Pops and returns the next entry from ` vertices_to_be_added ` that is not yet in the protocol state . Also returns a ` ProtocolOutcome ` that schedules the next action to add a vertex unless the queue is empty and ` ProtocolOutcome ` s to request missing dependencies .
pub(crate) fn pop_vertex_to_add(
        &mut self,
        highway: &Highway<C>,
        pending_values: &HashMap<ProposedBlock<C>, HashSet<(ValidVertex<C>, I)>>,
        max_requests_for_vertex: usize,
    ) -> (Option<PendingVertex<I, C>>, ProtocolOutcomes<I, C>) {
        let mut outcomes = Vec::new();
        // Get the next vertex to be added; skip the ones that are already in the protocol state,
        // and the ones that are still missing dependencies.
        loop {
            let pv = match self.vertices_no_deps.pop() {
                None => return (None, outcomes),
                Some(pv) if highway.has_vertex(pv.vertex()) => continue,
                Some(pv) => pv,
            };
            if let Some(dep) = highway.missing_dependency(pv.pvv()) {
                let sender = pv.sender().clone();
                let time_received = pv.time_received;
                // Find the first dependency that `pv` needs that we haven't synchronized yet
                // and request it from the sender of `pv`. Since it relies on it, it should have
                // it as well.
                let transitive_dependency =
                    self.find_transitive_dependency(dep.clone(), &sender, time_received);
                if self
                    .vertices_no_deps
                    .contains_dependency(&transitive_dependency)
                {
                    // `dep` is already downloaded and waiting in the synchronizer queue to be
                    // added, we don't have to request it again. Add the `pv`
                    // back to the queue so that it can be retried later. `dep` does not wait for
                    // any of the dependencies currently so it should be retried soon.
                    self.add_missing_dependency(dep.clone(), pv);
                    continue;
                }
                // We are still missing a dependency. Store the vertex in the map and request
                // the dependency from the sender.
                // Make `pv` depend on the direct dependency `dep` and not `transitive_dependency`
                // since there's a higher chance of adding `pv` to the protocol
                // state after `dep` is added, rather than `transitive_dependency`.
                self.add_missing_dependency(dep.clone(), pv);
                // If we already have the dependency and it is a proposal that is currently being
                // handled by the block validator, and this sender is already known as a source,
                // do nothing.
                if pending_values
                    .values()
                    .flatten()
                    .any(|(vv, s)| vv.inner().id() == transitive_dependency && s == &sender)
                {
                    continue;
                }
                // If we already have the dependency and it is a proposal that is currently being
                // handled by the block validator, and this sender is not yet known as a source,
                // we return the proposal as if this sender had sent it to us, so they get added.
                if let Some((vv, _)) = pending_values
                    .values()
                    .flatten()
                    .find(|(vv, _)| vv.inner().id() == transitive_dependency)
                {
                    debug!(
                        dependency = ?transitive_dependency, %sender,
                        "adding sender as a source for proposal"
                    );
                    let dep_pv = PendingVertex::new(sender, vv.clone().into(), time_received);
                    // We found the next vertex to add.
                    if !self.vertices_no_deps.is_empty() {
                        // There are still vertices in the queue: schedule next call.
                        outcomes.push(ProtocolOutcome::QueueAction(ACTION_ID_VERTEX));
                    }
                    return (Some(dep_pv), outcomes);
                }
                // If we have already requested the dependency from this peer, or from the maximum
                // number of peers, do nothing.
                let entry = self
                    .requests_sent
                    .entry(transitive_dependency.clone())
                    .or_default();
                if entry.len() >= max_requests_for_vertex || !entry.insert(sender.clone()) {
                    continue;
                }
                // Otherwise request the missing dependency from the sender.
                let uuid = thread_rng().next_u64();
                debug!(?uuid, dependency = ?transitive_dependency, %sender, "requesting dependency");
                let ser_msg =
                    HighwayMessage::RequestDependency(uuid, transitive_dependency).serialize();
                outcomes.push(ProtocolOutcome::CreatedTargetedMessage(ser_msg, sender));
                continue;
            }
            // We found the next vertex to add.
            if !self.vertices_no_deps.is_empty() {
                // There are still vertices in the queue: schedule next call.
                outcomes.push(ProtocolOutcome::QueueAction(ACTION_ID_VERTEX));
            }
            return (Some(pv), outcomes);
        }
    }
pub ( crate ) fn pop_vertex_to_add ( & mut self , highway : & Highway < C > , pending_values : & HashMap < ProposedBlock < C > , HashSet < ( ValidVertex < C > , I ) > > , max_requests_for_vertex : usize , ) -> ( Option < PendingVertex < I , C > > , ProtocolOutcomes < I , C > ) { let mut outcomes = Vec :: new ( ) ; // Get the next vertex to be added; skip the ones that are already in the protocol state, // and the ones that are still missing dependencies. loop { let pv = match self . vertices_no_deps . pop ( ) { None => return ( None , outcomes ) , Some ( pv ) if highway . has_vertex ( pv . vertex ( ) ) => continue , Some ( pv ) => pv , } ; if let Some ( dep ) = highway . missing_dependency ( pv . pvv ( ) ) { let sender = pv . sender ( ) . clone ( ) ; let time_received = pv . time_received ; // Find the first dependency that `pv` needs that we haven't synchronized yet // and request it from the sender of `pv`. Since it relies on it, it should have // it as well. let transitive_dependency = self . find_transitive_dependency ( dep . clone ( ) , & sender , time_received ) ; if self . vertices_no_deps . contains_dependency ( & transitive_dependency ) { // `dep` is already downloaded and waiting in the synchronizer queue to be // added, we don't have to request it again. Add the `pv` // back to the queue so that it can be retried later. `dep` does not wait for // any of the dependencies currently so it should be retried soon. self . add_missing_dependency ( dep . clone ( ) , pv ) ; continue ; } // We are still missing a dependency. Store the vertex in the map and request // the dependency from the sender. // Make `pv` depend on the direct dependency `dep` and not `transitive_dependency` // since there's a higher chance of adding `pv` to the protocol // state after `dep` is added, rather than `transitive_dependency`. self . add_missing_dependency ( dep . clone ( ) , pv ) ; // If we already have the dependency and it is a proposal that is currently being // handled by the block validator, and this sender is already known as a source, // do nothing. if pending_values . values ( ) . flatten ( ) . any ( | ( vv , s ) | vv . inner ( ) . id ( ) == transitive_dependency && s == & sender ) { continue ; } // If we already have the dependency and it is a proposal that is currently being // handled by the block validator, and this sender is not yet known as a source, // we return the proposal as if this sender had sent it to us, so they get added. if let Some ( ( vv , _ ) ) = pending_values . values ( ) . flatten ( ) . find ( | ( vv , _ ) | vv . inner ( ) . id ( ) == transitive_dependency ) { debug ! ( dependency transitive_dependency sender " " ) ; let dep_pv = PendingVertex :: new ( sender , vv . clone ( ) . into ( ) , time_received ) ; // We found the next vertex to add. if ! self . vertices_no_deps . is_empty ( ) { // There are still vertices in the queue: schedule next call. outcomes . push ( ProtocolOutcome :: QueueAction ( ACTION_ID_VERTEX ) ) ; } return ( Some ( dep_pv ) , outcomes ) ; } // If we have already requested the dependency from this peer, or from the maximum // number of peers, do nothing. let entry = self . requests_sent . entry ( transitive_dependency . clone ( ) ) . or_default ( ) ; if entry . len ( ) >= max_requests_for_vertex || ! entry . insert ( sender . clone ( ) ) { continue ; } // Otherwise request the missing dependency from the sender. let uuid = thread_rng ( ) . next_u64 ( ) ; debug ! ( uuid dependency transitive_dependency sender " " ) ; let ser_msg = HighwayMessage :: RequestDependency ( uuid , transitive_dependency ) . serialize ( ) ; outcomes . push ( ProtocolOutcome :: CreatedTargetedMessage ( ser_msg , sender ) ) ; continue ; } // We found the next vertex to add. if ! self . vertices_no_deps . is_empty ( ) { // There are still vertices in the queue: schedule next call. outcomes . push ( ProtocolOutcome :: QueueAction ( ACTION_ID_VERTEX ) ) ; } return ( Some ( pv ) , outcomes ) ; } }
Pops and returns the next entry from `vertices_to_be_added` that is not yet in the protocol
state.
Pops and returns the next entry from ` vertices_to_be_added ` that is not yet in the protocol state .
&mut self
highway
&Highway<C>
pending_values
&HashMap<ProposedBlock<C>, HashSet<(ValidVertex<C>, I)>>
max_requests_for_vertex
usize
&mut self
highway
&Highway<C>
pending_values
&HashMap<ProposedBlock<C>, HashSet<(ValidVertex<C>, I)>>
max_requests_for_vertex
usize