Backport #954 to v2.3: Fix shred retransmit (#955)

Author: buffalu

turbine/src/retransmit_stage.rs

@@ -411,15 +411,8 @@ fn retransmit_shred(
         return None;
     }
     let mut compute_turbine_peers = Measure::start("turbine_start");
-    let (root_distance, addrs) = get_retransmit_addrs(
-        &key,
-        root_bank,
-        cache,
-        addr_cache,
-        socket_addr_space,
-        stats,
-        shred_receiver_addr,
-    )?;
+    let (root_distance, addrs) =
+        get_retransmit_addrs(&key, root_bank, cache, addr_cache, socket_addr_space, stats)?;
     compute_turbine_peers.stop();
     stats
         .compute_turbine_peers_total
@@ -441,7 +434,13 @@ fn retransmit_shred(
             RetransmitSocket::Xdp(sender) => {
                 let mut sent = num_addrs;
                 if num_addrs > 0 {
-                    if let Err(e) = sender.try_send(key.index(), addrs.to_vec(), shred) {
+                    // shred receiver not included in the stats
+                    let mut send_addrs = Vec::with_capacity(num_addrs + 1);
+                    send_addrs.extend(addrs.iter());
+                    if let Some(addr) = shred_receiver_addr {
+                        send_addrs.push(*addr);
+                    }
+                    if let Err(e) = sender.try_send(key.index(), send_addrs, shred) {
                         log::warn!("xdp channel full: {e:?}");
                         stats
                             .num_shreds_dropped_xdp_full
@@ -451,13 +450,24 @@ fn retransmit_shred(
                 }
                 sent
             }
-            RetransmitSocket::Socket(socket) => match multi_target_send(socket, shred, &addrs) {
-                Ok(()) => num_addrs,
-                Err(SendPktsError::IoError(ioerr, num_failed)) => {
-                    error!("retransmit_to multi_target_send error: {ioerr:?}, {num_failed}/{} packets failed", num_addrs);
-                    num_addrs - num_failed
+            RetransmitSocket::Socket(socket) => {
+                let mut send_addrs = Vec::with_capacity(num_addrs + 1);
+                send_addrs.extend(addrs.iter());
+                // shred receiver not included in the stats
+                if let Some(addr) = shred_receiver_addr {
+                    send_addrs.push(*addr);
+                }
+                match multi_target_send(socket, shred, &send_addrs) {
+                    Ok(()) => num_addrs,
+                    Err(SendPktsError::IoError(ioerr, num_failed)) => {
+                        error!(
+                            "retransmit_to multi_target_send error: {ioerr:?}, \
+                         {num_failed}/{num_addrs} packets failed"
+                        );
+                        num_addrs.saturating_sub(num_failed)
+                    }
                 }
-            },
+            }
         },
     };
     retransmit_time.stop();
@@ -487,25 +497,21 @@ fn get_retransmit_addrs<'a>(
     addr_cache: &'a AddrCache,
     socket_addr_space: &SocketAddrSpace,
     stats: &RetransmitStats,
-    shred_receiver_addr: &Option<SocketAddr>,
 ) -> Option<(/*root_distance:*/ u8, Cow<'a, [SocketAddr]>)> {
     if let Some((root_distance, addrs)) = addr_cache.get(shred) {
         stats.addr_cache_hit.fetch_add(1, Ordering::Relaxed);
         return Some((root_distance, Cow::Borrowed(addrs)));
     }
     let (slot_leader, cluster_nodes) = cache.get(&shred.slot())?;
     let data_plane_fanout = cluster_nodes::get_data_plane_fanout(shred.slot(), root_bank);
-    let (root_distance, mut addrs) = cluster_nodes
+    let (root_distance, addrs) = cluster_nodes
         .get_retransmit_addrs(slot_leader, shred, data_plane_fanout, socket_addr_space)
         .inspect_err(|err| match err {
             Error::Loopback { .. } => {
                 stats.num_loopback_errs.fetch_add(1, Ordering::Relaxed);
             }
         })
         .ok()?;
-    if let Some(shred_receiver_addr) = shred_receiver_addr {
-        addrs.push(*shred_receiver_addr);
-    }
     stats.addr_cache_miss.fetch_add(1, Ordering::Relaxed);
     Some((root_distance, Cow::Owned(addrs)))
 }