packet recv allocations

Cargo.toml

@@ -92,6 +92,7 @@ hdrhistogram = "7.5.4"
 hex = { version = "0.4", features = ["serde"] }
 hkdf = "0.12"
 libc = "0.2"
+mimalloc = "0.1.52"
 mio = { version = "1.0.4", features = ["net", "os-poll"] }
 pprof = { version = "0.13", features = ["criterion", "flamegraph"] }
 proc-macro2 = "1"

crates/bin/Cargo.toml

@@ -16,6 +16,7 @@ silver_peer.workspace = true
 
 flux.workspace = true
 hex.workspace = true
+mimalloc.workspace = true
 quinn-proto.workspace = true
 rand.workspace = true
 tracing.workspace = true

crates/bin/src/main.rs

@@ -15,6 +15,10 @@ use silver_network::{Context, NetworkTile, P2p};
 use silver_peer::PeerManager;
 use tracing_subscriber::{EnvFilter, fmt::format::FmtSpan};
 
+#[cfg(not(feature = "alloc-profile"))]
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
 fn main() -> Result<(), Box<dyn Error>> {
     #[cfg(feature = "alloc-profile")]
     let _alloc_profile_guard = silver_common::allocator::init_allocator_trace();

crates/e2e/Cargo.toml

@@ -9,6 +9,7 @@ version.workspace = true
 buffa.workspace = true
 flux.workspace = true
 hdrhistogram.workspace = true
+mimalloc.workspace = true
 quinn-proto.workspace = true
 rand.workspace = true
 silver_beacon_state.workspace = true
@@ -41,6 +42,7 @@ tokio = { version = "1", features = ["rt", "time", "macros"], optional = true }
 [features]
 default = []
 lh-client = ["dep:futures", "dep:libp2p", "dep:tokio", "dep:async-trait"]
+alloc-profile = ["silver_common/alloc-profile"]
 
 [dev-dependencies]
 tracing.workspace = true

crates/e2e/examples/gossip_oneway.rs

@@ -38,6 +38,7 @@ use std::{
 };
 
 use flux::{tile::Tile, timing::Nanos};
+use mimalloc::MiMalloc;
 use rand::{Rng, RngCore};
 use silver_common::{GossipMsgOut, GossipTopic, NewGossipMsg, P2pSend, PeerEvent, TRandomAccess};
 use silver_e2e::{
@@ -55,7 +56,14 @@ const FORK_DIGEST_HEX: &str = "abcd1234";
 const TOPIC: GossipTopic = GossipTopic::BeaconBlock;
 const DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
 
+#[cfg(not(feature = "alloc-profile"))]
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
 fn main() {
+    #[cfg(feature = "alloc-profile")]
+    let _alloc_profile_guard = silver_common::allocator::init_allocator_trace();
+
     tracing_subscriber::fmt().with_max_level(tracing::Level::WARN).try_init().ok();
     let args = parse_args();
     assert!(args.payload_size >= 8, "payload-size must be >= 8 for the timestamp prefix");

crates/e2e/examples/gossip_oneway_lh.rs

@@ -44,6 +44,9 @@ const TOPIC: GossipTopic = GossipTopic::BeaconBlock;
 const DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
 
 fn main() {
+    #[cfg(feature = "alloc-profile")]
+    let _alloc_profile_guard = silver_common::allocator::init_allocator_trace();
+
     tracing_subscriber::fmt().with_max_level(tracing::Level::WARN).try_init().ok();
     let args = parse_args();
     assert!(args.payload_size >= 8, "payload-size must be >= 8 for the timestamp prefix");

crates/network/src/socket/linux.rs

@@ -71,6 +71,7 @@ impl RxBatch {
         // `self.count` weren't `take()`d last round — they still have
         // valid iovec pointers and RX_BUF_SIZE len.
         for i in 0..self.count {
+            self.bufs[i].clear();
             if !self.bufs[i].try_reclaim(RX_BUF_SIZE) {
                 self.bufs[i] = BytesMut::with_capacity(RX_BUF_SIZE);
             }

crates/network/src/socket/portable.rs

@@ -28,6 +28,7 @@ impl RxBatch {
     pub(crate) fn recv(&mut self, socket: &UdpSocket) -> usize {
         // Reclaim only slots that were used last round.
         for &(buf_idx, _len, _addr) in &self.datagrams {
+            self.bufs[buf_idx].clear();
             if !self.bufs[buf_idx].try_reclaim(RX_BUF_SIZE) {
                 self.bufs[buf_idx] = BytesMut::with_capacity(RX_BUF_SIZE);
             }

crates/network/src/tile.rs

@@ -263,14 +263,6 @@ where
         }
 
         let now = Instant::now();
-        p2p::p2p_spin(
-            &self.poll,
-            &mut self.p2p_endpoint,
-            &mut self.p2p_socket,
-            &mut self.context,
-            now,
-            &mut |evt| on_event(Event::P2pNet(evt)),
-        );
 
         for evt in &self.events {
             if evt.token() == DISC_SOCKET_TOKEN && evt.is_readable() {
@@ -287,6 +279,15 @@ where
             }
         }
 
+        p2p::p2p_spin(
+            &self.poll,
+            &mut self.p2p_endpoint,
+            &mut self.p2p_socket,
+            &mut self.context,
+            now,
+            &mut |evt| on_event(Event::P2pNet(evt)),
+        );
+
         self.disc_socket.flush(&self.poll);
         self.discovery.poll(|disc_event| match disc_event {
             DiscoveryEvent::SendMessage { to, data } => {