This commit is contained in:
Mateusz Gruszczyński committed 2026-11-07 08:16:11 +01:00
1 parent c038ad44b3
commit cd3c28d390
33 files changed
+807 -221

No files matched your search

+5 -5
View File
@@ -277,8 +277,8 @@ struct ColdEntry {
#[derive(Default)]
struct ColdStore {
entries: HashMap<Key, ColdEntry>,
variants: HashMap<Key, Vec<Key>>,
entries: KeyMap<ColdEntry>,
variants: KeyMap<Vec<Key>>,
lru: BTreeMap<u64, Key>,
bytes: u64,
sequence: u64,
@@ -304,8 +304,8 @@ enum ColdCommand {
}
#[derive(Default)]
struct Store {
entries: HashMap<Key, Entry>,
variants: HashMap<Key, Vec<Key>>,
entries: KeyMap<Entry>,
variants: KeyMap<Vec<Key>>,
lru: BTreeMap<u64, Key>,
bytes: usize,
logical_bytes: usize,
@@ -314,7 +314,7 @@ struct Store {
revision: u64,
control_revision: u64,
scope_generations: HashMap<String, u64>,
key_generations: HashMap<Key, u64>,
key_generations: KeyMap<u64>,
}
impl Store {
fn index_variant(&mut self, base_key: Key, key: Key) {
+32 -24
View File
@@ -268,13 +268,12 @@ impl Settings {
if !self.enabled {
return;
}
if let Some(value) = &self.via_header {
headers.append(
http::header::VIA,
value
.parse::<HeaderValue>()
.expect("validated cache Via header"),
);
if let Some(value) = self
.via_header
.as_deref()
.and_then(crate::settings::cached_header_value)
{
headers.append(http::header::VIA, value);
}
if self.expose_only_hits && outcome != "HIT" {
return;
@@ -300,7 +299,7 @@ impl Settings {
headers,
"x-cache-hits",
"x-upstream-cache-hits",
HeaderValue::from_str(&hits.to_string()).expect("integer header"),
HeaderValue::from(hits),
);
}
if self.expose_cache_status {
@@ -311,30 +310,33 @@ impl Settings {
)
.then_some(("stale", Some(StatusCode::NOT_MODIFIED)))
});
let mut value = if let Some((why, status)) = forwarded {
let mut value = format!("RustProxy; fwd={why}");
use std::fmt::Write as _;
let mut value = String::with_capacity(96);
value.push_str("RustProxy; ");
if let Some((why, status)) = forwarded {
let _ = write!(value, "fwd={why}");
if let Some(status) = status {
value.push_str(&format!("; fwd-status={}", status.as_u16()));
let _ = write!(value, "; fwd-status={}", status.as_u16());
}
value
} else {
match outcome {
"HIT" => format!("RustProxy; hit; ttl={ttl}"),
"MISS" => "RustProxy; fwd=miss".to_owned(),
"BYPASS" if reason == Some("request_method") => {
"RustProxy; fwd=method".to_owned()
"HIT" => {
let _ = write!(value, "hit; ttl={ttl}");
}
_ => "RustProxy; fwd=bypass".to_owned(),
"MISS" => value.push_str("fwd=miss"),
"BYPASS" if reason == Some("request_method") => value.push_str("fwd=method"),
_ => value.push_str("fwd=bypass"),
}
};
}
if let Some(reason) = reason {
value.push_str(&format!("; detail=\"{reason}\""));
let _ = write!(value, "; detail=\"{reason}\"");
}
if transformed {
value.push_str("; transformed");
}
headers.append(::http::header::CACHE_STATUS,
HeaderValue::from_str(&value).expect("generated Cache-Status member"),
HeaderValue::from_maybe_shared(bytes::Bytes::from(value))
.expect("generated Cache-Status member"),
);
}
}
@@ -345,7 +347,7 @@ impl Settings {
headers,
"x-cache-resident-age",
"x-upstream-cache-resident-age",
HeaderValue::from_str(&resident_age.to_string()).expect("integer header"),
HeaderValue::from(resident_age),
);
}
}
@@ -357,9 +359,15 @@ fn insert_cache_diagnostic(
upstream_name: &'static str,
value: HeaderValue,
) {
let upstream_values: Vec<HeaderValue> = headers.get_all(name).iter().cloned().collect();
for upstream_value in upstream_values {
headers.append(upstream_name, upstream_value);
let name = http::HeaderName::from_static(name);
// Fast path: the origin did not send this diagnostic, so there is nothing to
// preserve under the X-Upstream-* alias and no temporary vector is needed.
if headers.contains_key(&name) {
let upstream_values: Vec<HeaderValue> = headers.get_all(&name).iter().cloned().collect();
let upstream_name = http::HeaderName::from_static(upstream_name);
for upstream_value in upstream_values {
headers.append(&upstream_name, upstream_value);
}
}
headers.insert(name, value);
}
+69
View File
@@ -28,6 +28,36 @@ const COLD_QUEUE: usize = 16;
const COLD_FORMAT_VERSION: u8 = 3;
type Key = [u8; 32];
/// Cache keys are SHA-256 digests, so their bytes are already uniformly
/// distributed and not attacker-steerable. Hashing them again with SipHash
/// costs measurable time on every HOT lookup; this hasher folds the digest
/// cheaply instead (any other input is still mixed, Fx-style).
#[derive(Default, Clone, Copy)]
struct DigestKeyHasher(u64);
impl std::hash::Hasher for DigestKeyHasher {
fn finish(&self) -> u64 {
self.0
}
fn write(&mut self, bytes: &[u8]) {
if bytes.len() == 32 {
let mut word = [0u8; 8];
word.copy_from_slice(&bytes[..8]);
self.0 ^= u64::from_le_bytes(word);
return;
}
for chunk in bytes.chunks(8) {
let mut word = [0u8; 8];
word[..chunk.len()].copy_from_slice(chunk);
self.0 = (self.0.rotate_left(5) ^ u64::from_le_bytes(word))
.wrapping_mul(0x51_7c_c1_b7_27_22_0a_95);
}
}
fn write_usize(&mut self, _length: usize) {
// The only usize written for `[u8; 32]` is its constant length prefix.
}
}
type KeyMap<V> = HashMap<Key, V, std::hash::BuildHasherDefault<DigestKeyHasher>>;
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
enum CleanupPriority {
Automatic(u64, SystemTime, SystemTime),
@@ -41,7 +71,46 @@ struct CleanupCandidate {
key: Key,
}
/// Minimum spacing between two `malloc_trim` calls. Trimming walks every glibc
/// arena under its lock and can take milliseconds with a large heap; doing it
/// after each eviction made a full HOT cache with constant turnover several
/// times slower. Skipped requests are remembered and flushed by the cache
/// maintenance tick, so freed memory still returns to the OS within seconds.
const ALLOCATOR_RELEASE_INTERVAL_MS: u64 = 250;
static ALLOCATOR_RELEASE_LAST_MS: AtomicU64 = AtomicU64::new(0);
static ALLOCATOR_RELEASE_PENDING: std::sync::atomic::AtomicBool =
std::sync::atomic::AtomicBool::new(false);
fn allocator_clock_ms() -> u64 {
static EPOCH: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
// +1 so the very first request is never mistaken for "just trimmed".
(EPOCH.get_or_init(Instant::now).elapsed().as_millis() as u64)
.saturating_add(ALLOCATOR_RELEASE_INTERVAL_MS + 1)
}
fn release_allocator_memory() {
let now = allocator_clock_ms();
let last = ALLOCATOR_RELEASE_LAST_MS.load(Ordering::Relaxed);
if now.saturating_sub(last) < ALLOCATOR_RELEASE_INTERVAL_MS
|| ALLOCATOR_RELEASE_LAST_MS
.compare_exchange(last, now, Ordering::AcqRel, Ordering::Relaxed)
.is_err()
{
ALLOCATOR_RELEASE_PENDING.store(true, Ordering::Relaxed);
return;
}
ALLOCATOR_RELEASE_PENDING.store(false, Ordering::Relaxed);
cache_trim_allocator();
}
/// Called from the periodic cache maintenance task.
fn flush_pending_allocator_release() {
if ALLOCATOR_RELEASE_PENDING.load(Ordering::Relaxed) {
release_allocator_memory();
}
}
fn cache_trim_allocator() {
#[cfg(all(target_os = "linux", target_env = "gnu"))]
unsafe {
libc::malloc_trim(0);