[CL-129] partial payloads: segmented gossip send path

crates/bin/src/main.rs

@@ -217,6 +217,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     );
     let identify = config.identify()?;
     let p2p_context = Context {
+        data_columns_consumer: None,
         gossip_producer: incoming_gossip_producer,
         gossip_consumer: outgoing_gossip_producer
             .cache_ref()

crates/common/src/spine/messages.rs

@@ -8,8 +8,8 @@ use flux::timing::Nanos;
 use silver_beacon_state_data::SLOTS_PER_EPOCH;
 
 use crate::{
-    DataKind, Enr, GossipTopic, Identify, MessageId, Origin, P2pStreamId, PeerId, StreamProtocol,
-    TCacheProducer, TCacheRead, TMultiProducer,
+    CacheFrameRef, DataKind, Enr, GossipTopic, Identify, MessageId, Origin, P2pStreamId, PeerId,
+    StreamProtocol, TCacheProducer, TCacheRead, TMultiProducer,
     column_util::columns_of,
     ssz_view::{
         BLOCKS_BY_RANGE_REQ_SIZE, DC_BY_RANGE_REQ_MAX,
@@ -711,6 +711,7 @@ pub enum RpcSeverity {
 #[allow(clippy::large_enum_variant)]
 pub enum P2pSend {
     Gossip(GossipMsgOut),
+    SegmentedGossip { peer_id: usize, frame: CacheFrameRef },
     Identify(usize),
     Rpc(RpcOutbound),
 }
@@ -719,14 +720,15 @@ impl P2pSend {
     pub fn peer_id(&self) -> usize {
         match self {
             P2pSend::Gossip(gossip_msg_out) => gossip_msg_out.peer_id,
+            P2pSend::SegmentedGossip { peer_id, .. } => *peer_id,
             P2pSend::Identify(peer) => *peer,
             P2pSend::Rpc(rpc_outbound) => rpc_outbound.peer_id(),
         }
     }
 
     pub fn protocol(&self) -> StreamProtocol {
         match self {
-            P2pSend::Gossip(_) => StreamProtocol::GossipSub,
+            P2pSend::Gossip(_) | P2pSend::SegmentedGossip { .. } => StreamProtocol::GossipSub,
             P2pSend::Identify(_) => StreamProtocol::Identity,
             P2pSend::Rpc(rpc_outbound) => rpc_outbound.protocol(),
         }

crates/e2e/src/stack.rs

@@ -241,6 +241,7 @@ impl PublisherStack {
             .expect("cluster outbound random access");
 
         let context = Context {
+            data_columns_consumer: None,
             gossip_producer: gossip_in_producer,
             gossip_consumer: gossip_out_ra_for_network,
             rpc_producer: rpc_in_producer,
@@ -373,6 +374,7 @@ impl EchoStack {
             .expect("cluster outbound random access");
 
         let context = Context {
+            data_columns_consumer: None,
             gossip_producer: gossip_in_producer,
             gossip_consumer: protobuf_ra_for_network,
             rpc_producer: rpc_in_producer,

crates/network/benches/quic_basic.rs

@@ -67,6 +67,7 @@ pub fn broadcast(c: &mut Criterion) {
                             );
 
                             let context = Context {
+                                data_columns_consumer: None,
                                 gossip_producer: gi_producer,
                                 gossip_consumer: go_consumer,
                                 rpc_producer: rpc_in,
@@ -133,6 +134,7 @@ pub fn broadcast(c: &mut Criterion) {
                                 cluster_in.cache_ref().random_access("cluster_out", true).unwrap();
 
                             let context = Context {
+                                data_columns_consumer: None,
                                 gossip_producer: gi_producer,
                                 gossip_consumer: go_consumer,
                                 rpc_producer: rpc_in,

crates/network/benches/quic_pingpong.rs

@@ -66,6 +66,7 @@ pub fn broadcast(c: &mut Criterion) {
                         let p2p = P2p::new(keypair, server_endpoint, 1024, Default::default());
 
                         let context = Context {
+                            data_columns_consumer: None,
                             gossip_producer: gi_producer,
                             gossip_consumer: go_consumer,
                             rpc_producer: rpc_in,
@@ -139,6 +140,7 @@ pub fn broadcast(c: &mut Criterion) {
                         cluster_in.cache_ref().random_access("cluster_out", true).unwrap();
 
                     let context = Context {
+                        data_columns_consumer: None,
                         gossip_producer: gi_producer,
                         gossip_consumer: go_consumer,
                         rpc_producer: rpc_in,

crates/network/src/lib.rs

@@ -41,6 +41,16 @@ silver_common::declare_counters! {
         GossipStallDisconnect,
         // RPC codecs currently retained in the network-tile-wide free list.
         RpcCodecPoolIdle,
+        CacheSegmentedAdmitted,
+        CacheSegmentedRejected,
+        CacheSegmentedCapacity,
+        CacheSegmentedSegments,
+        CacheSegmentedOwnerAllocations,
+        CacheSegmentedFrames,
+        // Reserved owner slots, including owners already held by Quinn.
+        CacheSegmentedOwners,
+        // Reserved ranges per recipient, including descriptors; not unique cache backing bytes.
+        CacheSegmentedRetainedBytes,
     }
 }
 

crates/network/src/p2p/context.rs

@@ -15,6 +15,7 @@ pub struct Context {
     pub cluster_nodes: Option<ClusterNodes>,
     pub cluster_inbound_producer: TProducer,
     pub cluster_outbound_consumer: TRandomAccess,
+    pub data_columns_consumer: Option<Box<TRandomAccess>>,
 }
 
 impl Context {

crates/network/src/p2p/mod.rs

@@ -13,12 +13,13 @@ use buffa::{Message, MessageView};
 pub use context::{ClusterNodes, Context};
 use fxhash::{FxHashMap, FxHashSet};
 use mio::{Poll, net::UdpSocket};
+use quic::SegmentedGossipLimits;
 pub(crate) use quic::{Peer, create_client_config};
 pub use quic::{SendResult, create_endpoint, create_server_config};
 use quinn_proto::{ConnectionHandle, DatagramEvent, Endpoint};
 use silver_common::{
-    ClusterMsgOut, GossipMsgOut, Identify, Keypair, P2pConnectionStats, P2pStreamId, PeerId,
-    ProtoIdentify, ProtoIdentifyView, RpcOutbound, RpcRequestOutbound, TCacheRead,
+    CacheFrameRef, ClusterMsgOut, GossipMsgOut, Identify, Keypair, P2pConnectionStats, P2pStreamId,
+    PeerId, ProtoIdentify, ProtoIdentifyView, RpcOutbound, RpcRequestOutbound, TCacheRead,
 };
 
 use crate::{
@@ -112,6 +113,8 @@ pub struct P2p {
     keypair: Keypair,
     endpoint: Endpoint,
     peers: FxHashMap<ConnectionHandle, Peer>,
+    // Peer queues and Quinn owners must drop before their shared budgets.
+    segmented_limits: Option<Box<SegmentedGossipLimits>>,
     rpc_codec_pool: RpcCodecPool,
     banned: FxHashSet<PeerId>,
     timeout: Option<Duration>,
@@ -134,6 +137,7 @@ impl P2p {
             keypair,
             endpoint,
             peers: FxHashMap::default(),
+            segmented_limits: None,
             rpc_codec_pool: RpcCodecPool::default(),
             banned: FxHashSet::default(),
             timeout: Some(Duration::ZERO),
@@ -358,6 +362,9 @@ impl P2p {
         };
 
         NetworkCounters::P2pConnections.set(self.peers.len() as u64);
+        if let Some(limits) = &self.segmented_limits {
+            limits.publish_gauges();
+        }
         did_work
     }
 
@@ -371,6 +378,23 @@ impl P2p {
         }
     }
 
+    pub fn enqueue_segmented_gossip(
+        &mut self,
+        peer_id: usize,
+        frame: CacheFrameRef,
+        context: &mut Context,
+    ) -> SendResult {
+        match self.peers.get_mut(&ConnectionHandle(peer_id)) {
+            Some(peer) => peer.send_segmented_gossip(
+                frame,
+                context,
+                self.segmented_limits.get_or_insert_with(Box::default),
+                &mut self.rpc_codec_pool,
+            ),
+            None => SendResult::UnknownPeer,
+        }
+    }
+
     pub fn enqueue_rpc_out(&mut self, msg: RpcOutbound, context: &mut Context) -> SendResult {
         match self.peers.get_mut(&ConnectionHandle(msg.peer_id())) {
             Some(peer) => {

crates/network/src/p2p/quic/gossip_frame.rs

@@ -0,0 +1,195 @@
+use std::{cell::Cell, ptr::NonNull, time::Instant};
+
+use bytes::Bytes;
+use silver_common::{
+    AcquiredCacheFrame, AcquiredCacheSegment, AcquiredRange, CacheFrameView, TRead,
+};
+
+use super::{Leased, leased::OutboundLeaseWheel};
+use crate::{NetworkCounters, p2p::Context};
+
+const MAX_RETAINED_BYTES: usize = 64 * 1024 * 1024;
+const MAX_RETAINED_OWNERS: usize = 8 * 1024;
+
+#[derive(Debug)]
+pub(crate) enum OutboundGossip {
+    Contiguous(Leased<TRead>),
+    Segmented(SegmentedFrame),
+}
+
+pub(crate) struct SegmentedGossipLimits {
+    frames: Cell<usize>,
+    owners: Cell<usize>,
+    retained_bytes: Cell<usize>,
+    max_frames: usize,
+    max_bytes: usize,
+    max_owners: usize,
+}
+
+impl Default for SegmentedGossipLimits {
+    fn default() -> Self {
+        Self::new(128)
+    }
+}
+
+impl SegmentedGossipLimits {
+    pub(crate) fn new(max_frames: usize) -> Self {
+        Self {
+            frames: Cell::new(0),
+            owners: Cell::new(0),
+            retained_bytes: Cell::new(0),
+            max_frames,
+            max_bytes: MAX_RETAINED_BYTES,
+            max_owners: MAX_RETAINED_OWNERS,
+        }
+    }
+
+    pub(crate) fn acquire(
+        &self,
+        view: CacheFrameView,
+        context: &mut Context,
+        wheel: &OutboundLeaseWheel,
+        now: Instant,
+    ) -> Option<SegmentedFrame> {
+        let bytes = view.descriptor_len().checked_add(view.wire_len())?;
+        let owners = view.segment_count() + 1;
+        if bytes > self.max_bytes.saturating_sub(self.retained_bytes.get()) ||
+            owners > self.max_owners.saturating_sub(self.owners.get()) ||
+            self.frames.get() >= self.max_frames
+        {
+            NetworkCounters::CacheSegmentedCapacity.inc();
+            return None;
+        }
+        self.frames.set(self.frames.get() + 1);
+        self.owners.set(self.owners.get() + owners);
+        self.retained_bytes.set(self.retained_bytes.get() + bytes);
+        let budget = FrameBudget { limits: NonNull::from(self), owners, bytes };
+        let frame = view.acquire_segments(
+            &mut context.gossip_consumer,
+            context.data_columns_consumer.as_deref_mut(),
+        )?;
+        NetworkCounters::CacheSegmentedAdmitted.inc();
+        NetworkCounters::CacheSegmentedSegments.add(frame.segment_count() as u64);
+        Some(SegmentedFrame { segments: wheel.leased(frame, now), budget })
+    }
+
+    pub(crate) fn publish_gauges(&self) {
+        NetworkCounters::CacheSegmentedFrames.set(self.frames.get() as u64);
+        NetworkCounters::CacheSegmentedOwners.set(self.owners.get() as u64);
+        NetworkCounters::CacheSegmentedRetainedBytes.set(self.retained_bytes.get() as u64);
+    }
+}
+
+impl Drop for SegmentedGossipLimits {
+    fn drop(&mut self) {
+        debug_assert_eq!(self.frames.get(), 0, "limits dropped with active frames");
+        debug_assert_eq!(self.owners.get(), 0, "limits dropped with active owners");
+        debug_assert_eq!(self.retained_bytes.get(), 0, "limits dropped with retained bytes");
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct SegmentedFrame {
+    segments: Leased<AcquiredCacheFrame>,
+    budget: FrameBudget,
+}
+
+impl SegmentedFrame {
+    pub(crate) fn wire_len(&self) -> usize {
+        self.segments.wire_len()
+    }
+
+    pub(crate) fn into_writer(self) -> SegmentedWriter {
+        let remaining = self.wire_len();
+        SegmentedWriter { frame: self, current: Bytes::new(), descriptor: Bytes::new(), remaining }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct SegmentedWriter {
+    frame: SegmentedFrame,
+    current: Bytes,
+    descriptor: Bytes,
+    remaining: usize,
+}
+
+impl SegmentedWriter {
+    pub(crate) fn chunk(&mut self) -> Option<&mut Bytes> {
+        if self.current.is_empty() {
+            let SegmentedFrame { segments, budget } = &mut self.frame;
+            self.current = match segments.take_next()? {
+                AcquiredCacheSegment::Framing(range) => {
+                    if self.descriptor.is_empty() {
+                        self.descriptor = budget.owner(segments.child(segments.descriptor_range()));
+                    }
+                    self.descriptor.slice(range)
+                }
+                AcquiredCacheSegment::Data(range) => budget.owner(segments.child(range)),
+            };
+        }
+        Some(&mut self.current)
+    }
+
+    pub(crate) fn written(&mut self, bytes: usize) -> bool {
+        assert!(bytes <= self.remaining);
+        self.remaining -= bytes;
+        self.remaining == 0
+    }
+}
+
+#[derive(Debug)]
+struct FrameBudget {
+    limits: NonNull<SegmentedGossipLimits>,
+    owners: usize,
+    bytes: usize,
+}
+
+// Budgets and owners remain on NetworkTile. P2p's boxed limits outlive all
+// peer queues, stream state, and Quinn-owned Bytes.
+unsafe impl Send for FrameBudget {}
+
+impl FrameBudget {
+    fn owner(&mut self, data: Leased<AcquiredRange>) -> Bytes {
+        assert!(self.owners > 0 && data.len() <= self.bytes);
+        self.owners -= 1;
+        self.bytes -= data.len();
+        let owner = SegmentOwner { data, limits: self.limits };
+        NetworkCounters::CacheSegmentedOwnerAllocations.inc();
+        Bytes::from_owner(owner)
+    }
+}
+
+impl Drop for FrameBudget {
+    fn drop(&mut self) {
+        let limits = unsafe { self.limits.as_ref() };
+        limits.frames.set(limits.frames.get() - 1);
+        limits.owners.set(limits.owners.get() - self.owners);
+        limits.retained_bytes.set(limits.retained_bytes.get() - self.bytes);
+    }
+}
+
+struct SegmentOwner {
+    data: Leased<AcquiredRange>,
+    limits: NonNull<SegmentedGossipLimits>,
+}
+
+// Bytes requires Send. Creation and destruction remain on NetworkTile,
+// whose boxed limits outlive every peer's Quinn connection.
+unsafe impl Send for SegmentOwner {}
+
+impl AsRef<[u8]> for SegmentOwner {
+    fn as_ref(&self) -> &[u8] {
+        self.data.as_ref()
+    }
+}
+
+impl Drop for SegmentOwner {
+    fn drop(&mut self) {
+        let limits = unsafe { self.limits.as_ref() };
+        limits.owners.set(limits.owners.get() - 1);
+        limits.retained_bytes.set(limits.retained_bytes.get() - self.data.len());
+    }
+}
+
+#[cfg(test)]
+mod tests;