pub fn retransmitter(
    sockets: Arc<Vec<UdpSocket>>,
    bank_forks: Arc<RwLock<BankForks>>,
    leader_schedule_cache: Arc<LeaderScheduleCache>,
    cluster_info: Arc<ClusterInfo>,
    shreds_receiver: mpsc::Receiver<Vec<Shred>>,
    max_slots: Arc<MaxSlots>,
    rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
) -> JoinHandle<()> {
    let cluster_nodes_cache = ClusterNodesCache::<RetransmitStage>::new(
        CLUSTER_NODES_CACHE_NUM_EPOCH_CAP,
        CLUSTER_NODES_CACHE_TTL,
    );
    let mut hasher_reset_ts = Instant::now();
    let mut stats = RetransmitStats::default();
    let shreds_received = Mutex::new((LruCache::new(DEFAULT_LRU_SIZE), PacketHasher::default()));
    let first_shreds_received = Mutex::<BTreeSet<Slot>>::default();
    let num_threads = get_thread_count().min(8).max(sockets.len());
    let thread_pool = ThreadPoolBuilder::new()
        .num_threads(num_threads)
        .thread_name(|i| format!("retransmit-{}", i))
        .build()
        .unwrap();
    Builder::new()
        .name("analog-retransmitter".to_string())
        .spawn(move || {
            trace!("retransmitter started");
            loop {
                match retransmit(
                    &thread_pool,
                    &bank_forks,
                    &leader_schedule_cache,
                    &cluster_info,
                    &shreds_receiver,
                    &sockets,
                    &mut stats,
                    &cluster_nodes_cache,
                    &mut hasher_reset_ts,
                    &shreds_received,
                    &max_slots,
                    &first_shreds_received,
                    rpc_subscriptions.as_deref(),
                ) {
                    Ok(()) => (),
                    Err(RecvTimeoutError::Timeout) => (),
                    Err(RecvTimeoutError::Disconnected) => break,
                }
            }
            trace!("exiting retransmitter");
        })
        .unwrap()
}
/// Service to retransmit messages from the leader or layer 1 to relevant peer nodes.
/// See `cluster_info` for network layer definitions.
/// # Arguments
/// * `sockets` - Sockets to read from.
/// * `bank_forks` - The BankForks structure
/// * `leader_schedule_cache` - The leader schedule to verify shreds
/// * `cluster_info` - This structure needs to be updated and populated by the bank and via gossip.
/// * `r` - Receive channel for shreds to be retransmitted to all the layer 1 nodes.
Service to retransmit messages from the leader or layer 1 to relevant peer nodes.
See `cluster_info` for network layer definitions.
Service to retransmit messages from the leader or layer 1 to relevant peer nodes . See ` cluster_info ` for network layer definitions .
pub fn retransmitter(
    sockets: Arc<Vec<UdpSocket>>,
    bank_forks: Arc<RwLock<BankForks>>,
    leader_schedule_cache: Arc<LeaderScheduleCache>,
    cluster_info: Arc<ClusterInfo>,
    shreds_receiver: mpsc::Receiver<Vec<Shred>>,
    max_slots: Arc<MaxSlots>,
    rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
) -> JoinHandle<()> {
    let cluster_nodes_cache = ClusterNodesCache::<RetransmitStage>::new(
        CLUSTER_NODES_CACHE_NUM_EPOCH_CAP,
        CLUSTER_NODES_CACHE_TTL,
    );
    let mut hasher_reset_ts = Instant::now();
    let mut stats = RetransmitStats::default();
    let shreds_received = Mutex::new((LruCache::new(DEFAULT_LRU_SIZE), PacketHasher::default()));
    let first_shreds_received = Mutex::<BTreeSet<Slot>>::default();
    let num_threads = get_thread_count().min(8).max(sockets.len());
    let thread_pool = ThreadPoolBuilder::new()
        .num_threads(num_threads)
        .thread_name(|i| format!("retransmit-{}", i))
        .build()
        .unwrap();
    Builder::new()
        .name("analog-retransmitter".to_string())
        .spawn(move || {
            trace!("retransmitter started");
            loop {
                match retransmit(
                    &thread_pool,
                    &bank_forks,
                    &leader_schedule_cache,
                    &cluster_info,
                    &shreds_receiver,
                    &sockets,
                    &mut stats,
                    &cluster_nodes_cache,
                    &mut hasher_reset_ts,
                    &shreds_received,
                    &max_slots,
                    &first_shreds_received,
                    rpc_subscriptions.as_deref(),
                ) {
                    Ok(()) => (),
                    Err(RecvTimeoutError::Timeout) => (),
                    Err(RecvTimeoutError::Disconnected) => break,
                }
            }
            trace!("exiting retransmitter");
        })
        .unwrap()
}
pub fn retransmitter ( sockets : Arc < Vec < UdpSocket > > , bank_forks : Arc < RwLock < BankForks > > , leader_schedule_cache : Arc < LeaderScheduleCache > , cluster_info : Arc < ClusterInfo > , shreds_receiver : mpsc :: Receiver < Vec < Shred > > , max_slots : Arc < MaxSlots > , rpc_subscriptions : Option < Arc < RpcSubscriptions > > , ) -> JoinHandle < ( ) > { let cluster_nodes_cache = ClusterNodesCache :: < RetransmitStage > :: new ( CLUSTER_NODES_CACHE_NUM_EPOCH_CAP , CLUSTER_NODES_CACHE_TTL , ) ; let mut hasher_reset_ts = Instant :: now ( ) ; let mut stats = RetransmitStats :: default ( ) ; let shreds_received = Mutex :: new ( ( LruCache :: new ( DEFAULT_LRU_SIZE ) , PacketHasher :: default ( ) ) ) ; let first_shreds_received = Mutex :: < BTreeSet < Slot > > :: default ( ) ; let num_threads = get_thread_count ( ) . min ( 8 ) . max ( sockets . len ( ) ) ; let thread_pool = ThreadPoolBuilder :: new ( ) . num_threads ( num_threads ) . thread_name ( | i | format ! ( " " i ) ) . build ( ) . unwrap ( ) ; Builder :: new ( ) . name ( " " . to_string ( ) ) . spawn ( move | | { trace ! ( " " ) ; loop { match retransmit ( & thread_pool , & bank_forks , & leader_schedule_cache , & cluster_info , & shreds_receiver , & sockets , & mut stats , & cluster_nodes_cache , & mut hasher_reset_ts , & shreds_received , & max_slots , & first_shreds_received , rpc_subscriptions . as_deref ( ) , ) { Ok ( ( ) ) => ( ) , Err ( RecvTimeoutError :: Timeout ) => ( ) , Err ( RecvTimeoutError :: Disconnected ) => break , } } trace ! ( " " ) ; } ) . unwrap ( ) }
Service to retransmit messages from the leader or layer 1 to relevant peer nodes.
Service to retransmit messages from the leader or layer 1 to relevant peer nodes .
sockets
Arc<Vec<UdpSocket>>
bank_forks
Arc<RwLock<BankForks>>
leader_schedule_cache
Arc<LeaderScheduleCache>
cluster_info
Arc<ClusterInfo>
shreds_receiver
mpsc::Receiver<Vec<Shred>>
max_slots
Arc<MaxSlots>
rpc_subscriptions
Option<Arc<RpcSubscriptions>>
sockets
Arc<Vec<UdpSocket>>
bank_forks
Arc<RwLock<BankForks>>
leader_schedule_cache
Arc<LeaderScheduleCache>
cluster_info
Arc<ClusterInfo>
shreds_receiver
mpsc::Receiver<Vec<Shred>>
max_slots
Arc<MaxSlots>
rpc_subscriptions
Option<Arc<RpcSubscriptions>>