Replace shared map with larger bitset

This simplifies the receiver dedup implementation by replacing it with
an inline bitset, which we shift over on new arrivals. The end result is
that we have an effective tracker for ~896 packets fitting into about
the same amount of space as the previous combination of shared map and
entry data (+6 bytes if we assume 500k entries).

The new implementation is also easier to tweak, since we directly expose
the window size. It's likely that this slows down inserts a little (we
need to shift over 112 bytes) but shuffling a 112-byte array should be
sufficiently fast that I don't think it's worth worrying about it.

dc/s2n-quic-dc/Cargo.toml

@@ -44,6 +44,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"], optional = tr
 zerocopy = { version = "0.7", features = ["derive"] }
 zeroize = "1"
 parking_lot = "0.12"
+bitvec = { version = "1.0.1", default-features = false }
 
 [dev-dependencies]
 bolero = "0.12"

dc/s2n-quic-dc/src/path/secret/map.rs

@@ -188,8 +188,6 @@ impl Map {
         let mut stateless_reset = [0; control::TAG_LEN];
         aws_lc_rs::rand::fill(&mut stateless_reset).unwrap();
 
-        let receiver_shared = receiver::Shared::new();
-
         let mut ids = Vec::with_capacity(peers.len());
         for (idx, (ciphersuite, version, peer)) in peers.into_iter().enumerate() {
             secret[..8].copy_from_slice(&(idx as u64).to_be_bytes()[..]);
@@ -206,7 +204,7 @@ impl Map {
                 peer,
                 secret,
                 sender,
-                receiver_shared.clone().new_receiver(),
+                receiver::State::new(),
                 dc::testing::TEST_APPLICATION_PARAMS,
                 dc::testing::TEST_REHANDSHAKE_PERIOD,
             );
@@ -226,7 +224,7 @@ impl Map {
     #[doc(hidden)]
     #[cfg(any(test, feature = "testing"))]
     pub fn test_insert(&self, peer: SocketAddr) {
-        let receiver = self.store.receiver().clone().new_receiver();
+        let receiver = super::receiver::State::new();
         let entry = Entry::fake(peer, Some(receiver));
         self.store.test_insert(entry);
     }
@@ -259,7 +257,7 @@ impl Map {
                 peer_addr,
                 secret,
                 sender,
-                map.store.receiver().clone().new_receiver(),
+                super::receiver::State::new(),
                 dc::testing::TEST_APPLICATION_PARAMS,
                 dc::testing::TEST_REHANDSHAKE_PERIOD,
             );

dc/s2n-quic-dc/src/path/secret/map/entry.rs

@@ -106,7 +106,7 @@ impl Entry {
 
     #[cfg(any(test, feature = "testing"))]
     pub fn fake(peer: SocketAddr, receiver: Option<receiver::State>) -> Arc<Entry> {
-        let receiver = receiver.unwrap_or_else(receiver::State::without_shared);
+        let receiver = receiver.unwrap_or_default();
 
         let mut secret = [0; 32];
         aws_lc_rs::rand::fill(&mut secret).unwrap();

dc/s2n-quic-dc/src/path/secret/map/entry/tests.rs

@@ -15,7 +15,7 @@ fn entry_size() {
     if should_check {
         assert_eq!(
             Entry::fake((std::net::Ipv4Addr::LOCALHOST, 0).into(), None).size(),
-            239
+            295
         );
     }
 }

dc/s2n-quic-dc/src/path/secret/map/handshake.rs

@@ -4,7 +4,7 @@
 use super::{Entry, Map};
 use crate::{
     packet::secret_control as control,
-    path::secret::{schedule, sender},
+    path::secret::{receiver, schedule, sender},
 };
 use s2n_quic_core::{
     dc::{self, ApplicationParams, DatagramInfo},
@@ -123,7 +123,7 @@ impl dc::Path for HandshakingPath {
                 .into_inner(),
         );
 
-        let receiver = self.map.store.receiver().clone().new_receiver();
+        let receiver = receiver::State::new();
 
         let entry = Entry::new(
             self.peer,

dc/s2n-quic-dc/src/path/secret/map/state.rs

@@ -78,8 +78,6 @@ where
     // FIXME: This will get replaced with sending on a handshake socket associated with the map.
     pub(super) control_socket: Arc<std::net::UdpSocket>,
 
-    pub(super) receiver_shared: Arc<receiver::Shared>,
-
     cleaner: Cleaner,
 
     init_time: Timestamp,
@@ -136,7 +134,6 @@ where
             requested_handshakes: Default::default(),
             cleaner: Cleaner::new(),
             signer,
-            receiver_shared: receiver::Shared::new(),
             control_socket,
             init_time,
             clock,
@@ -556,10 +553,6 @@ where
         &self.signer
     }
 
-    fn receiver(&self) -> &Arc<receiver::Shared> {
-        &self.receiver_shared
-    }
-
     fn send_control_packet(&self, dst: &SocketAddr, buffer: &mut [u8]) {
         match self.control_socket.send_to(buffer, dst) {
             Ok(_) => {

dc/s2n-quic-dc/src/path/secret/map/state/tests.rs

@@ -136,7 +136,7 @@ impl Model {
                     ip,
                     secret,
                     sender::State::new(stateless_reset),
-                    state.receiver().clone().new_receiver(),
+                    receiver::State::new(),
                     dc::testing::TEST_APPLICATION_PARAMS,
                     dc::testing::TEST_REHANDSHAKE_PERIOD,
                 )));

dc/s2n-quic-dc/src/path/secret/map/store.rs

@@ -43,8 +43,6 @@ pub trait Store: 'static + Send + Sync {
 
     fn signer(&self) -> &stateless_reset::Signer;
 
-    fn receiver(&self) -> &Arc<receiver::Shared>;
-
     fn send_control_packet(&self, dst: &SocketAddr, buffer: &mut [u8]);
 
     fn rehandshake_period(&self) -> Duration;