gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-10-12 15:59:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Replace dubious CidQueue::is_active_retired logic

Browse Source 
This predicate was based on whether an entry in the circular CID
buffer was unset. However, that entry might have already been
overwritten with newer data. Meanwhile, because we guarantee that the
active CID is always at the start of the sliding window of CID
sequence numbers, we can easily infer whether the previously-active
CID was retired and replaced by retire_prior_to by whether the retired
set it returns is empty.
This commit is contained in:
Benjamin Saunders
2022-02-08 18:19:05 -08:00
committed by Dirkjan Ochtman
parent 37337e3287
commit 1146db7356
2 changed files with 125 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quinn-proto/src/cid_queue.rs
+98 -56
View File
@@ -1,6 +1,6 @@
use std::ops::Range;
use crate::{shared::IssuedCid, ConnectionId, ResetToken};
use crate::{frame::NewConnectionId, ConnectionId, ResetToken};
/// DataType stored in CidQueue buffer
type CidData = (ConnectionId, Option<ResetToken>);
@@ -31,42 +31,65 @@ impl CidQueue {
}
}
pub fn insert(&mut self, cid: IssuedCid) -> Result<(), InsertError> {
if cid.sequence == self.offset && self.buffer[self.cursor].is_some() {
// This is a duplicate of the active CID.
return Ok(());
}
/// Handle a `NEW_CONNECTION_ID` frame
///
/// Returns a non-empty range of retired sequence numbers and the reset token of the new active
/// CID iff any CIDs were retired.
pub fn insert(
&mut self,
cid: NewConnectionId,
) -> Result<Option<(Range<u64>, ResetToken)>, InsertError> {
// Position of new CID wrt. the current active CID
let index = match cid.sequence.checked_sub(self.offset) {
None => return Err(InsertError::Retired),
Some(x) => x,
};
if index >= Self::LEN as u64 {
let retired_count = cid.retire_prior_to.saturating_sub(self.offset);
if index >= Self::LEN as u64 + retired_count {
return Err(InsertError::ExceedsLimit);
}
let index = (self.cursor + index as usize) % Self::LEN;
self.buffer[index] = Some((cid.id, Some(cid.reset_token)));
Ok(())
}
if index == 0 && self.buffer[self.cursor].is_some() {
// This is a duplicate of the active CID, so no action is necessary. Because we always
// retire CIDs preceding the active CID, retire_prior_to cannot prompt any action
// either.
return Ok(None);
}
/// Returns the possibly-empty range of newly retired CIDs
pub fn retire_prior_to(&mut self, sequence: u64) -> Range<u64> {
let n = match sequence.checked_sub(self.offset) {
None => return 0..0,
Some(n) => n as usize,
};
for i in 0..n {
// Discard retired CIDs, if any
for i in 0..(retired_count.min(Self::LEN as u64) as usize) {
self.buffer[(self.cursor + i) % Self::LEN] = None;
}
// Record the new CID
let index = ((self.cursor as u64 + index) % Self::LEN as u64) as usize;
self.buffer[index] = Some((cid.id, Some(cid.reset_token)));
if retired_count == 0 {
return Ok(None);
}
// The active CID was retired. Find the first known CID with sequence number of at least
// retire_prior_to, and inform the caller that all prior CIDs have been retired, and of
// the new CID's reset token.
self.cursor = ((self.cursor as u64 + retired_count) % Self::LEN as u64) as usize;
let (i, (_, token)) = self
.iter()
.next()
.expect("it is impossible to retire a CID without supplying a new one");
self.cursor = (self.cursor + i) % Self::LEN;
let orig_offset = self.offset;
self.offset = sequence;
self.cursor = (self.cursor + n) % Self::LEN;
orig_offset..sequence
self.offset = cid.retire_prior_to + i as u64;
Ok(Some((
orig_offset..self.offset,
token.expect("non-initial CID missing reset token"),
)))
}
/// Switch to next active CID if possible, return
/// 1) the corresponding ResetToken and 2) a possibly-empty range preceding it to retire
/// 1) the corresponding ResetToken and 2) a non-empty range preceding it to retire
pub fn next(&mut self) -> Option<(ResetToken, Range<u64>)> {
let (i, cid_data) = self.iter().next()?;
let (i, cid_data) = self.iter().nth(1)?;
self.buffer[self.cursor] = None;
let orig_offset = self.offset;
@@ -76,15 +99,16 @@ impl CidQueue {
Some((cid_data.1.unwrap(), orig_offset..sequence))
}
/// Iterate inactive CIDs in CidQueue that are not `None`
/// Iterate CIDs in CidQueue that are not `None`, including the active CID
fn iter(&self) -> impl Iterator<Item = (usize, CidData)> + '_ {
(1..Self::LEN).filter_map(move |step| {
(0..Self::LEN).filter_map(move |step| {
let index = (self.cursor + step) % Self::LEN;
self.buffer[index].map(|cid_data| (step, cid_data))
})
}
pub fn update_cid(&mut self, cid: ConnectionId) {
/// Replace the initial CID
pub fn update_initial_cid(&mut self, cid: ConnectionId) {
debug_assert_eq!(self.offset, 0);
self.buffer[self.cursor] = Some((cid, None));
}
@@ -94,11 +118,6 @@ impl CidQueue {
self.buffer[self.cursor].unwrap().0
}
/// Check whether self.offset points to a valid CID in CidQueue
pub fn is_active_retired(&mut self) -> bool {
self.buffer[self.cursor].is_none()
}
/// Return the sequence number of active remote CID
pub fn active_seq(&self) -> u64 {
self.offset
@@ -119,11 +138,12 @@ pub enum InsertError {
mod tests {
use super::*;
fn cid(sequence: u64) -> IssuedCid {
IssuedCid {
fn cid(sequence: u64, retire_prior_to: u64) -> NewConnectionId {
NewConnectionId {
sequence,
id: ConnectionId::new(&[0xAB; 8]),
reset_token: ResetToken::from([0xCD; crate::RESET_TOKEN_SIZE]),
retire_prior_to,
}
}
@@ -138,7 +158,7 @@ mod tests {
assert!(q.next().is_none());
for i in 1..CidQueue::LEN as u64 {
q.insert(cid(i)).unwrap();
q.insert(cid(i, 0)).unwrap();
}
for i in 1..CidQueue::LEN as u64 {
let (_, retire) = q.next().unwrap();
@@ -152,7 +172,7 @@ mod tests {
let mut q = CidQueue::new(initial_cid());
let seqs = (1..CidQueue::LEN as u64).filter(|x| x % 2 == 0);
for i in seqs.clone() {
q.insert(cid(i)).unwrap();
q.insert(cid(i, 0)).unwrap();
}
for i in seqs {
let (_, retire) = q.next().unwrap();
@@ -168,13 +188,13 @@ mod tests {
let mut q = CidQueue::new(initial_cid());
for i in 1..CidQueue::LEN as u64 {
q.insert(cid(i)).unwrap();
q.insert(cid(i, 0)).unwrap();
}
for _ in 1..(CidQueue::LEN as u64 - 1) {
q.next().unwrap();
}
for i in CidQueue::LEN as u64..(CidQueue::LEN as u64 + 3) {
q.insert(cid(i)).unwrap();
q.insert(cid(i, 0)).unwrap();
}
for i in (CidQueue::LEN as u64 - 1)..(CidQueue::LEN as u64 + 3) {
q.next().unwrap();
@@ -184,35 +204,53 @@ mod tests {
}
#[test]
fn retire() {
fn retire_dense() {
let mut q = CidQueue::new(initial_cid());
for i in 1..CidQueue::LEN as u64 {
q.insert(cid(i)).unwrap();
q.insert(cid(i, 0)).unwrap();
}
assert_eq!(q.active_seq(), 0);
assert_eq!(q.retire_prior_to(2), 0..2);
let r = q.retire_prior_to(2);
assert_eq!(r.end - r.start, 0);
assert_eq!(q.insert(cid(4, 2)).unwrap().unwrap().0, 0..2);
assert_eq!(q.active_seq(), 2);
assert_eq!(q.insert(cid(4, 2)), Ok(None));
for i in 2..(CidQueue::LEN as u64 - 1) {
let _ = q.next().unwrap();
assert_eq!(q.active_seq(), i + 1);
let retire = q.retire_prior_to(i + 1);
assert_eq!(retire.end - retire.start, 0);
assert!(!q.is_active_retired());
assert_eq!(q.insert(cid(i + 1, i + 1)), Ok(None));
}
assert!(q.next().is_none());
assert!(!q.is_active_retired());
}
#[test]
fn retire_sparse() {
// Retiring CID 0 when CID 1 is not known should retire CID 1 as we move to CID 2
let mut q = CidQueue::new(initial_cid());
q.insert(cid(2, 0)).unwrap();
assert_eq!(q.insert(cid(3, 1)).unwrap().unwrap().0, 0..2,);
assert_eq!(q.active_seq(), 2);
}
#[test]
fn retire_many() {
let mut q = CidQueue::new(initial_cid());
q.insert(cid(2, 0)).unwrap();
assert_eq!(
q.insert(cid(1_000_000, 1_000_000)).unwrap().unwrap().0,
0..1_000_000,
);
assert_eq!(q.active_seq(), 1_000_000);
}
#[test]
fn insert_limit() {
let mut q = CidQueue::new(initial_cid());
assert_eq!(q.insert(cid(CidQueue::LEN as u64 - 1)), Ok(()));
assert_eq!(q.insert(cid(CidQueue::LEN as u64 - 1, 0)), Ok(None));
assert_eq!(
q.insert(cid(CidQueue::LEN as u64)),
q.insert(cid(CidQueue::LEN as u64, 0)),
Err(InsertError::ExceedsLimit)
);
}
@@ -220,19 +258,23 @@ mod tests {
#[test]
fn insert_duplicate() {
let mut q = CidQueue::new(initial_cid());
q.insert(cid(0)).unwrap();
q.insert(cid(0)).unwrap();
q.insert(cid(0, 0)).unwrap();
q.insert(cid(0, 0)).unwrap();
}
#[test]
fn insert_retired() {
let mut q = CidQueue::new(initial_cid());
assert_eq!(q.insert(cid(0)), Ok(()), "reinserting active CID succeeds");
assert_eq!(
q.insert(cid(0, 0)),
Ok(None),
"reinserting active CID succeeds"
);
assert!(q.next().is_none(), "active CID isn't requeued");
q.insert(cid(1)).unwrap();
q.insert(cid(1, 0)).unwrap();
q.next().unwrap();
assert_eq!(
q.insert(cid(0)),
q.insert(cid(0, 0)),
Err(InsertError::Retired),
"previous active CID is already retired"
);
@@ -242,12 +284,12 @@ mod tests {
fn retire_then_insert_next() {
let mut q = CidQueue::new(initial_cid());
for i in 1..CidQueue::LEN as u64 {
q.insert(cid(i)).unwrap();
q.insert(cid(i, 0)).unwrap();
}
q.next().unwrap();
q.insert(cid(CidQueue::LEN as u64)).unwrap();
q.insert(cid(CidQueue::LEN as u64, 0)).unwrap();
assert_eq!(
q.insert(cid(CidQueue::LEN as u64 + 1)),
q.insert(cid(CidQueue::LEN as u64 + 1, 0)),
Err(InsertError::ExceedsLimit)
);
}
quinn-proto/src/connection/mod.rs
+27 -31
View File
@@ -26,8 +26,9 @@ use crate::{
range_set::ArrayRangeSet,
shared::{
ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent,
EndpointEventInner, IssuedCid,
EndpointEventInner,
},
token::ResetToken,
transport_parameters::TransportParameters,
Dir, Frame, Side, StreamId, Transmit, TransportError, TransportErrorCode, VarInt,
MAX_STREAM_COUNT, MIN_INITIAL_SIZE, RESET_TOKEN_SIZE, TIMER_GRANULARITY,
@@ -2022,7 +2023,7 @@ impl Connection {
trace!("retrying with CID {}", rem_cid);
let client_hello = state.client_hello.take().unwrap();
self.retry_src_cid = Some(rem_cid);
self.rem_cids.update_cid(rem_cid);
self.rem_cids.update_initial_cid(rem_cid);
self.rem_handshake_cid = rem_cid;
let space = &mut self.spaces[SpaceId::Initial];
@@ -2146,7 +2147,7 @@ impl Connection {
if !state.rem_cid_set {
trace!("switching remote CID to {}", rem_cid);
let mut state = state.clone();
self.rem_cids.update_cid(rem_cid);
self.rem_cids.update_initial_cid(rem_cid);
self.rem_handshake_cid = rem_cid;
self.orig_rem_cid = rem_cid;
state.rem_cid_set = true;
@@ -2469,45 +2470,36 @@ impl Connection {
));
}
let retired = self.rem_cids.retire_prior_to(frame.retire_prior_to);
self.spaces[SpaceId::Data]
.pending
.retire_cids
.extend(retired);
use crate::cid_queue::InsertError;
let new_rem_cid = IssuedCid {
sequence: frame.sequence,
id: frame.id,
reset_token: frame.reset_token,
};
match self.rem_cids.insert(new_rem_cid) {
Ok(()) => {}
match self.rem_cids.insert(frame) {
Ok(None) => {}
Ok(Some((retired, reset_token))) => {
self.spaces[SpaceId::Data]
.pending
.retire_cids
.extend(retired);
self.set_reset_token(reset_token);
}
Err(InsertError::ExceedsLimit) => {
return Err(TransportError::CONNECTION_ID_LIMIT_ERROR(""));
}
Err(InsertError::Retired) => {
trace!("discarding already-retired");
// RETIRE_CONNECTION_ID might not have been previously sent if e.g. a
// range of connection IDs larger than the active connection ID limit
// was retired all at once via retire_prior_to.
self.spaces[SpaceId::Data]
.pending
.retire_cids
.push(frame.sequence);
continue;
}
}
};
if self.side.is_server() && self.peer_params.stateless_reset_token.is_none() {
if self.rem_cids.active_seq() == 0 {
// We're a server still using the initial remote CID for the client, so
// let's switch immediately to enable clientside stateless resets.
let _ = self.update_rem_cid();
}
} else if self.rem_cids.is_active_retired() {
// If our current CID is meant to be retired; or
// active remote CID is invalid (due to packet loss or reordering),
// we must switch to next valid CID
self.update_rem_cid().unwrap();
if self.side.is_server() && self.rem_cids.active_seq() == 0 {
// We're a server still using the initial remote CID for the client, so
// let's switch immediately to enable clientside stateless resets.
let _ = self.update_rem_cid();
}
}
Frame::NewToken { token } => {
@@ -2619,7 +2611,11 @@ impl Connection {
// Retire the current remote CID and any CIDs we had to skip.
let retire_cids = &mut self.spaces[SpaceId::Data].pending.retire_cids;
retire_cids.extend(retired);
self.set_reset_token(reset_token);
Ok(())
}
fn set_reset_token(&mut self, reset_token: ResetToken) {
self.endpoint_events
.push_back(EndpointEventInner::ResetToken(
self.path.remote,
@@ -2629,7 +2625,6 @@ impl Connection {
// Reduce linkability
self.spin = false;
Ok(())
}
/// Issue an initial set of connection IDs to the peer
@@ -2881,10 +2876,11 @@ impl Connection {
(Some(x), y) => Some(cmp::min(x, y)),
};
if let Some(ref info) = params.preferred_address {
self.rem_cids.insert(IssuedCid {
self.rem_cids.insert(frame::NewConnectionId {
sequence: 1,
id: info.connection_id,
reset_token: info.stateless_reset_token,
retire_prior_to: 0,
}).expect("preferred address CID is the first received, and hence is guaranteed to be legal");
}
self.peer_params = params;