diff options
| -rw-r--r-- | src/access/screen.rs | 13 | ||||
| -rw-r--r-- | src/document/signatures.rs | 65 |
2 files changed, 69 insertions, 9 deletions
diff --git a/src/access/screen.rs b/src/access/screen.rs index a9450a6..4c640b6 100644 --- a/src/access/screen.rs +++ b/src/access/screen.rs @@ -39,7 +39,7 @@ use crate::document::replica::{Epoch, user_of}; use crate::document::signatures::Signatures; use crate::document::{Change, ChangeHash, Document, ReadDoc}; -/// Changes waiting for their dependencies, kept at most. +/// Changes waiting for their dependencies, kept at most: the oldest go. const MAX_WAITING: usize = 10_000; /// Refused changes kept for operators, at most: the oldest go. pub(crate) const MAX_REFUSED: usize = 1_000; @@ -187,6 +187,13 @@ impl Waiting { pub(crate) fn len(&self) -> usize { self.0.len() } + + /// The newest `MAX_WAITING`: the oldest go, so changes a peer sent that will never + /// be taken (whose dependencies or signatures never come) make way for others. + fn keep_newest(&mut self) { + let excess = self.0.len().saturating_sub(MAX_WAITING); + self.0.drain(..excess); + } } /// What `apply` did: how many changes it applied, and those it refused. @@ -269,7 +276,7 @@ pub(crate) fn apply( let (ready, unready) = in_order(doc, candidates); waiting.0 = unready; if ready.is_empty() { - waiting.0.truncate(MAX_WAITING); + waiting.keep_newest(); return Ok(screened); } @@ -462,7 +469,7 @@ pub(crate) fn apply( screened.roots.insert(root); } } - waiting.0.truncate(MAX_WAITING); + waiting.keep_newest(); Ok(screened) } diff --git a/src/document/signatures.rs b/src/document/signatures.rs index 919e7b7..27bbb9e 100644 --- a/src/document/signatures.rs +++ b/src/document/signatures.rs @@ -19,7 +19,7 @@ //! latest one a node verified, those a peer wasn't sent yet (`newer_than`). A received //! signature is checked once its change is there (`offer`), and kept if it vouches for //! more of its actor's changes than any before; one of a change not there yet waits -//! (`pending`, a bounded number). +//! (`pending`, a bounded number: the oldest go). use std::collections::HashMap; @@ -193,9 +193,47 @@ pub(crate) fn unpack(bytes: &[u8]) -> Option<Vec<(ChangeHash, [u8; 64])>> { .collect(), ) } -/// Received signatures of changes not there yet, kept at most (`offer`). +/// Received signatures of changes not there yet, kept at most (`offer`): the oldest go. const MAX_PENDING: usize = 4096; +/// Received signatures of changes the document doesn't hold yet, `MAX_PENDING` at most: +/// the oldest go, so junk a peer sent once (signatures of changes that never come) +/// makes way for those that keep coming. +#[derive(Debug, Default, Clone)] +struct Pending { + by_change: HashMap<ChangeHash, [u8; 64]>, + /// The changes, oldest first; some may have left `by_change` since. + order: std::collections::VecDeque<ChangeHash>, +} + +impl Pending { + fn insert(&mut self, hash: ChangeHash, signature: [u8; 64]) { + if self.by_change.insert(hash, signature).is_none() { + self.order.push_back(hash); + } + while self.by_change.len() > MAX_PENDING { + match self.order.pop_front() { + Some(oldest) => { + self.by_change.remove(&oldest); + } + None => break, + } + } + if self.order.len() > 2 * MAX_PENDING { + let by_change = &self.by_change; + self.order.retain(|hash| by_change.contains_key(hash)); + } + } + + fn remove(&mut self, hash: &ChangeHash) -> Option<[u8; 64]> { + self.by_change.remove(hash) + } + + fn keys(&self) -> impl Iterator<Item = &ChangeHash> { + self.by_change.keys() + } +} + /// A document's signatures, by change. #[derive(Debug, Default)] pub(crate) struct Signatures { @@ -211,7 +249,7 @@ pub(crate) struct Signatures { /// (`newer_than`). stamp: u64, /// Received signatures of changes the document doesn't hold yet. - pending: HashMap<ChangeHash, [u8; 64]>, + pending: Pending, } /// An actor's latest change a signature verified. @@ -316,9 +354,7 @@ impl Signatures { let mut kept = 0; for (hash, signature) in received { if doc.get_change_meta_by_hash(&hash).is_none() { - if self.pending.len() < MAX_PENDING { - self.pending.insert(hash, signature); - } + self.pending.insert(hash, signature); } else if self.vouch(doc, hash, &signature) { self.insert(hash, signature); kept += 1; @@ -636,6 +672,23 @@ mod tests { assert_eq!(kept.since(0), vec![signed(second)]); } + /// Signatures waiting for their changes: past the bound, the oldest go. + #[test] + fn the_oldest_waiting_signatures_go() { + let mut pending = Pending::default(); + let hash = |i: usize| { + let mut bytes = [0; 32]; + bytes[..8].copy_from_slice(&(i as u64).to_le_bytes()); + ChangeHash(bytes) + }; + for i in 0..=MAX_PENDING { + pending.insert(hash(i), [0; 64]); + } + assert_eq!(pending.keys().count(), MAX_PENDING); + assert!(pending.remove(&hash(0)).is_none()); + assert!(pending.remove(&hash(MAX_PENDING)).is_some()); + } + /// Protocol data, listed paths (within or above them) and asked-for writes are /// signed; others aren't. #[test] |
