aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--src/access/screen.rs13
-rw-r--r--src/document/signatures.rs65
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]