fn broadcast_single_peer<V>(peer: PeerNetworkId, backoff: bool, smp: &mut SharedMempool<V>) -> bool
where
    V: TransactionValidation,
{
    let peer_manager = &smp.peer_manager;

    let (timeline_id, retry_txns_id, next_backoff) = if peer_manager.is_picked_peer(&peer) {
        let state = peer_manager.get_peer_state(&peer);
        let next_backoff = state.broadcast_info.backoff_mode;
        if state.is_alive {
            (
                state.timeline_id,
                state
                    .broadcast_info
                    .total_retry_txns
                    .into_iter()
                    .collect::<Vec<_>>(),
                next_backoff,
            )
        } else {
            return next_backoff;
        }
    } else {
        return false;
    };

    // It is possible that a broadcast was scheduled as non-backoff before an ACK received after the
    // broadcast scheduling turns on backoff mode
    // If this is the case, ignore this schedule and wait till next broadcast scheduled as backoff
    if !backoff && next_backoff {
        return next_backoff;
    }

    // craft batch of txns to broadcast
    let mut mempool = smp
        .mempool
        .lock()
        .expect("[shared mempool] failed to acquire mempool lock");

    // first populate batch with retriable txns, to prioritize resending them
    let retry_txns = mempool.filter_read_timeline(retry_txns_id);
    // pad the batch with new txns from fresh timeline read, if batch has space
    let (new_txns, new_timeline_id) = if retry_txns.len() < smp.config.shared_mempool_batch_size {
        mempool.read_timeline(
            timeline_id,
            smp.config.shared_mempool_batch_size - retry_txns.len(),
        )
    } else {
        (vec![], timeline_id)
    };

    if new_txns.is_empty() && retry_txns.is_empty() {
        return next_backoff;
    }

    // read first tx in timeline
    let earliest_timeline_id = mempool
        .read_timeline(0, 1)
        .0
        .get(0)
        .expect("empty timeline")
        .0;
    // don't hold mempool lock during network send
    drop(mempool);

    // combine retry_txns and new_txns into batch
    let mut all_txns = retry_txns
        .into_iter()
        .chain(new_txns.into_iter())
        .collect::<Vec<_>>();
    all_txns.truncate(smp.config.shared_mempool_batch_size);
    let batch_timeline_ids = all_txns.iter().map(|(id, _txn)| *id).collect::<Vec<_>>();
    let batch_txns = all_txns
        .into_iter()
        .map(|(_id, txn)| txn)
        .collect::<Vec<_>>();

    let mut network_sender = smp
        .network_senders
        .get_mut(&peer.network_id())
        .expect("[shared mempool] missing network sender");

    let request_id = create_request_id(timeline_id, new_timeline_id);
    let txns_ct = batch_txns.len();
    if let Err(e) = send_mempool_sync_msg(
        MempoolSyncMsg::BroadcastTransactionsRequest {
            request_id: request_id.clone(),
            transactions: batch_txns,
        },
        peer.peer_id(),
        &mut network_sender,
    ) {
        error!(
            "[shared mempool] error broadcasting transactions to peer {:?}: {}",
            peer, e
        );
    } else {
        counters::SHARED_MEMPOOL_TRANSACTION_BROADCAST.inc_by(txns_ct as i64);
        peer_manager.update_peer_broadcast(
            peer,
            request_id,
            batch_timeline_ids,
            new_timeline_id,
            earliest_timeline_id,
        );
        notify_subscribers(SharedMempoolNotification::Broadcast, &smp.subscribers);
    }

    next_backoff
}
/// broadcasts txns to `peer` if alive
/// Returns whether the next broadcast scheduled for this peer should be in backpressure mode or not
broadcasts txns to `peer` if alive
Returns whether the next broadcast scheduled for this peer should be in backpressure mode or not
broadcasts txns to ` peer ` if alive Returns whether the next broadcast scheduled for this peer should be in backpressure mode or not
fn broadcast_single_peer<V>(peer: PeerNetworkId, backoff: bool, smp: &mut SharedMempool<V>) -> bool
where
    V: TransactionValidation,
{
    let peer_manager = &smp.peer_manager;
    let (timeline_id, retry_txns_id, next_backoff) = if peer_manager.is_picked_peer(&peer) {
        let state = peer_manager.get_peer_state(&peer);
        let next_backoff = state.broadcast_info.backoff_mode;
        if state.is_alive {
            (
                state.timeline_id,
                state
                    .broadcast_info
                    .total_retry_txns
                    .into_iter()
                    .collect::<Vec<_>>(),
                next_backoff,
            )
        } else {
            return next_backoff;
        }
    } else {
        return false;
    };
    // It is possible that a broadcast was scheduled as non-backoff before an ACK received after the
    // broadcast scheduling turns on backoff mode
    // If this is the case, ignore this schedule and wait till next broadcast scheduled as backoff
    if !backoff && next_backoff {
        return next_backoff;
    }
    // craft batch of txns to broadcast
    let mut mempool = smp
        .mempool
        .lock()
        .expect("[shared mempool] failed to acquire mempool lock");
    // first populate batch with retriable txns, to prioritize resending them
    let retry_txns = mempool.filter_read_timeline(retry_txns_id);
    // pad the batch with new txns from fresh timeline read, if batch has space
    let (new_txns, new_timeline_id) = if retry_txns.len() < smp.config.shared_mempool_batch_size {
        mempool.read_timeline(
            timeline_id,
            smp.config.shared_mempool_batch_size - retry_txns.len(),
        )
    } else {
        (vec![], timeline_id)
    };
    if new_txns.is_empty() && retry_txns.is_empty() {
        return next_backoff;
    }
    // read first tx in timeline
    let earliest_timeline_id = mempool
        .read_timeline(0, 1)
        .0
        .get(0)
        .expect("empty timeline")
        .0;
    // don't hold mempool lock during network send
    drop(mempool);
    // combine retry_txns and new_txns into batch
    let mut all_txns = retry_txns
        .into_iter()
        .chain(new_txns.into_iter())
        .collect::<Vec<_>>();
    all_txns.truncate(smp.config.shared_mempool_batch_size);
    let batch_timeline_ids = all_txns.iter().map(|(id, _txn)| *id).collect::<Vec<_>>();
    let batch_txns = all_txns
        .into_iter()
        .map(|(_id, txn)| txn)
        .collect::<Vec<_>>();
    let mut network_sender = smp
        .network_senders
        .get_mut(&peer.network_id())
        .expect("[shared mempool] missing network sender");
    let request_id = create_request_id(timeline_id, new_timeline_id);
    let txns_ct = batch_txns.len();
    if let Err(e) = send_mempool_sync_msg(
        MempoolSyncMsg::BroadcastTransactionsRequest {
            request_id: request_id.clone(),
            transactions: batch_txns,
        },
        peer.peer_id(),
        &mut network_sender,
    ) {
        error!(
            "[shared mempool] error broadcasting transactions to peer {:?}: {}",
            peer, e
        );
    } else {
        counters::SHARED_MEMPOOL_TRANSACTION_BROADCAST.inc_by(txns_ct as i64);
        peer_manager.update_peer_broadcast(
            peer,
            request_id,
            batch_timeline_ids,
            new_timeline_id,
            earliest_timeline_id,
        );
        notify_subscribers(SharedMempoolNotification::Broadcast, &smp.subscribers);
    }
    next_backoff
}
fn broadcast_single_peer < V > ( peer : PeerNetworkId , backoff : bool , smp : & mut SharedMempool < V > ) -> bool where V : TransactionValidation , { let peer_manager = & smp . peer_manager ; let ( timeline_id , retry_txns_id , next_backoff ) = if peer_manager . is_picked_peer ( & peer ) { let state = peer_manager . get_peer_state ( & peer ) ; let next_backoff = state . broadcast_info . backoff_mode ; if state . is_alive { ( state . timeline_id , state . broadcast_info . total_retry_txns . into_iter ( ) . collect :: < Vec < _ > > ( ) , next_backoff , ) } else { return next_backoff ; } } else { return false ; } ; // It is possible that a broadcast was scheduled as non-backoff before an ACK received after the // broadcast scheduling turns on backoff mode // If this is the case, ignore this schedule and wait till next broadcast scheduled as backoff if ! backoff && next_backoff { return next_backoff ; } // craft batch of txns to broadcast let mut mempool = smp . mempool . lock ( ) . expect ( " " ) ; // first populate batch with retriable txns, to prioritize resending them let retry_txns = mempool . filter_read_timeline ( retry_txns_id ) ; // pad the batch with new txns from fresh timeline read, if batch has space let ( new_txns , new_timeline_id ) = if retry_txns . len ( ) < smp . config . shared_mempool_batch_size { mempool . read_timeline ( timeline_id , smp . config . shared_mempool_batch_size - retry_txns . len ( ) , ) } else { ( vec ! [ ] , timeline_id ) } ; if new_txns . is_empty ( ) && retry_txns . is_empty ( ) { return next_backoff ; } // read first tx in timeline let earliest_timeline_id = mempool . read_timeline ( 0 , 1 ) . 0 . get ( 0 ) . expect ( " " ) . 0 ; // don't hold mempool lock during network send drop ( mempool ) ; // combine retry_txns and new_txns into batch let mut all_txns = retry_txns . into_iter ( ) . chain ( new_txns . into_iter ( ) ) . collect :: < Vec < _ > > ( ) ; all_txns . truncate ( smp . config . shared_mempool_batch_size ) ; let batch_timeline_ids = all_txns . iter ( ) . map ( | ( id , _txn ) | * id ) . collect :: < Vec < _ > > ( ) ; let batch_txns = all_txns . into_iter ( ) . map ( | ( _id , txn ) | txn ) . collect :: < Vec < _ > > ( ) ; let mut network_sender = smp . network_senders . get_mut ( & peer . network_id ( ) ) . expect ( " " ) ; let request_id = create_request_id ( timeline_id , new_timeline_id ) ; let txns_ct = batch_txns . len ( ) ; if let Err ( e ) = send_mempool_sync_msg ( MempoolSyncMsg :: BroadcastTransactionsRequest { request_id : request_id . clone ( ) , transactions : batch_txns , } , peer . peer_id ( ) , & mut network_sender , ) { error ! ( " " peer e ) ; } else { counters :: SHARED_MEMPOOL_TRANSACTION_BROADCAST . inc_by ( txns_ct as i64 ) ; peer_manager . update_peer_broadcast ( peer , request_id , batch_timeline_ids , new_timeline_id , earliest_timeline_id , ) ; notify_subscribers ( SharedMempoolNotification :: Broadcast , & smp . subscribers ) ; } next_backoff }
broadcasts txns to `peer` if alive
Returns whether the next broadcast scheduled for this peer should be in backpressure mode or not
broadcasts txns to ` peer ` if alive Returns whether the next broadcast scheduled for this peer should be in backpressure mode or not
peer
PeerNetworkId
backoff
bool
smp
&mut SharedMempool<V>
peer
PeerNetworkId
backoff
bool
smp
&mut SharedMempool<V>