Skip to content

Frame a block, and decode one, with one copy of the payload #192

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
matrixarkai merged 1 commit into from
Oct 24, 2026
Merged

Frame a block, and decode one, with one copy of the payload #192

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 73 additions & 7 deletions examples/allocation_report.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,6 +280,40 @@ fn main() {
// assumed from the fixture. Checked below.
let promote_disk_hits = promoting.stats().disk_hits;

// The same read asked for eight keys at once. The batch path reads the
// blocks together, decodes each one once, and hands the bytes to every
// position that asked for that key -- so what it must not do is a copy per
// position on top of the one each position gets.
//
// The groups are built outside the measured closure, as the resident batch
// row builds its key list outside: a batch row that formats eight keys a
// call measures `format!`.
let promote_groups: Vec<Vec<CacheKey>> = (0..PROMOTE_KEYS / BATCH)
.map(|group| {
((group * BATCH)..(group * BATCH + BATCH))
.map(key)
.collect()
})
.collect();
let mut promote_group_cursor = 0usize;
let batch_promotion_hits_before = promoting.stats().disk_hits;
rows.push(measure(
"get_batch, promoting from a lower tier",
BATCH,
|| {
promote_group_cursor += 1;
let group = &promote_groups[promote_group_cursor % promote_groups.len()];
let found = promoting
.get_batch(group)
.expect("a batch read of keys the fixture wrote must succeed");
assert!(
found.iter().all(|slot| slot.is_some()),
"the fixture has to still hold every key of the group"
);
},
));
let batch_promotion_hits = promoting.stats().disk_hits - batch_promotion_hits_before;

let mut reaching_ssd = 2_000_000usize;
rows.push(measure("put, reaching the SSD tier", 1, || {
reaching_ssd += 1;
Expand Down Expand Up @@ -385,6 +419,7 @@ fn main() {
// pinned batch of 8 15 -> 7 allocations
// a write that evicts 11.48 -> 5.93 allocations, 44,140.8 -> 19,537.3 bytes
// a batch of 8 shared 2 allocations for 8 keys
// a promoting read 13.00 -> 12.00 allocations, 25,334.0 -> 24,310.0 bytes
//
// Raising one is a deliberate act, and the number that replaced it goes
// in the table beside it.
Expand All @@ -397,7 +432,7 @@ fn main() {
("acquire_batch, 2 duplicated", 9.5, 4_300.0),
("put, overwriting", 2.5, 2_200.0),
("put, evicting", 6.5, 20_500.0),
("get, promoting from a lower tier", 15.0, 27_500.0),
("get, promoting from a lower tier", 13.0, 25_500.0),
] {
match find(path) {
Some(row) if row.allocations <= max_allocations && row.bytes <= max_bytes => {}
Expand All @@ -424,18 +459,39 @@ fn main() {
// RocksDB before 16.11 58,193.6
// RocksDB after 12.11 55,951.6
//
// And again when framing a block stopped copying the payload into a buffer
// of its own first:
//
// backend allocations bytes
// Rust after 27.11 58,223.8
// RocksDB after 11.11 54,914.2
//
// Per arm, because the two are not comparable: this allocator sees Rust
// only, and with RocksDB the store's own bytes are allocated elsewhere.
// Three runs of the same binary moved the figures by 0.01 allocations and
// 8 bytes, so the margin here is wide rather than lucky.
#[cfg(not(feature = "rocksdb-ssd"))]
const SSD_WRITE_ALLOCATIONS: f64 = 30.0;
const BATCH_PROMOTION_ALLOCATIONS: f64 = 65.0;
#[cfg(feature = "rocksdb-ssd")]
const BATCH_PROMOTION_ALLOCATIONS: f64 = 80.0;
match find("get_batch, promoting from a lower tier") {
Some(row) if row.allocations <= BATCH_PROMOTION_ALLOCATIONS && row.bytes <= 190_000.0 => {}
Some(row) => failures.push(format!(
"a batch of {BATCH} promoting reads allocates {:.2} times and {:.1} bytes, against \
{BATCH_PROMOTION_ALLOCATIONS:.0} and 190,000",
row.allocations, row.bytes
)),
None => failures.push("the promoting batch read is missing from the table".to_string()),
}

#[cfg(not(feature = "rocksdb-ssd"))]
const SSD_WRITE_BYTES: f64 = 60_500.0;
const SSD_WRITE_ALLOCATIONS: f64 = 28.5;
#[cfg(not(feature = "rocksdb-ssd"))]
const SSD_WRITE_BYTES: f64 = 59_000.0;
#[cfg(feature = "rocksdb-ssd")]
const SSD_WRITE_ALLOCATIONS: f64 = 14.0;
const SSD_WRITE_ALLOCATIONS: f64 = 12.5;
#[cfg(feature = "rocksdb-ssd")]
const SSD_WRITE_BYTES: f64 = 57_000.0;
const SSD_WRITE_BYTES: f64 = 55_700.0;
match find("put, reaching the SSD tier") {
Some(row) if row.allocations <= SSD_WRITE_ALLOCATIONS && row.bytes <= SSD_WRITE_BYTES => {}
Some(row) => failures.push(format!(
Expand Down Expand Up @@ -475,9 +531,19 @@ fn main() {
// a larger memory capacity, a placement change -- and the row would go on
// reporting the cost of a memory hit under the promotion label.
let promote_reads = REPEATS as u64 + 1;
if promote_disk_hits + 16 < promote_reads {
if promote_disk_hits * 10 < promote_reads * 9 {
failures.push(format!(
"the promoting row made {promote_reads} reads and only {promote_disk_hits} of them reached a lower tier, so it is not measuring a promotion"
));
}

// The same check for the batch row, over its own reads: eight keys asked
// for on each of two hundred and one calls, all but a handful of which
// have to come off the SSD tier.
let batch_promotion_reads = (REPEATS as u64 + 1) * BATCH as u64;
if batch_promotion_hits * 10 < batch_promotion_reads * 9 {
failures.push(format!(
"the promoting row made {promote_reads} reads and only {promote_disk_hits} of them reached a lower tier, so it is not measuring a promotion"
"the promoting batch row asked for {batch_promotion_reads} keys and only {batch_promotion_hits} came off a lower tier"
));
}

Expand Down
165 changes: 152 additions & 13 deletions src/runtime/builder_and_gc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,29 +287,42 @@ const CACHE_CODEC_RAW: u8 = 0;
const CACHE_CODEC_ZSTD: u8 = 1;
const CACHE_HEADER_LEN: usize = 8 + 1 + 1 + 8 + 8;

/// Frame a value as a block, with one copy of the payload.
///
/// Every arm of this used to `value.to_vec()` the payload into an owned buffer
/// and then copy that buffer into the block -- two payload-sized allocations
/// and two copies per block written, where the block itself is the only buffer
/// anyone keeps. Three of the four arms store the value raw, so three of the
/// four were paying for a copy that existed only to be copied again.
///
/// The compressed payload still needs a buffer of its own, because whether it
/// is used at all is decided by how big it turns out to be. The raw payload
/// does not: it is already in the caller's slice.
fn encode_cache_block(value: &[u8], options: CacheBlockOptions) -> Result<Vec<u8>, CacheError> {
let (codec, payload) = match options.compression {
CacheCompression::None if value.len() >= options.min_compress_bytes => {
(CACHE_CODEC_RAW, value.to_vec())
}
CacheCompression::None => (CACHE_CODEC_RAW, value.to_vec()),
let compressed = match options.compression {
CacheCompression::Zstd { level } if value.len() >= options.min_compress_bytes => {
let compressed = zstd::stream::encode_all(value, level)?;
// Kept only if framing it is smaller than storing the value as it
// is, which is the same comparison as before.
if CACHE_HEADER_LEN + compressed.len() < value.len() {
(CACHE_CODEC_ZSTD, compressed)
Some(compressed)
} else {
(CACHE_CODEC_RAW, value.to_vec())
None
}
}
CacheCompression::Zstd { .. } => (CACHE_CODEC_RAW, value.to_vec()),
CacheCompression::Zstd { .. } | CacheCompression::None => None,
};
let (codec, payload): (u8, &[u8]) = match compressed.as_deref() {
Some(compressed) => (CACHE_CODEC_ZSTD, compressed),
None => (CACHE_CODEC_RAW, value),
};
let mut block = Vec::with_capacity(CACHE_HEADER_LEN + payload.len());
block.extend_from_slice(CACHE_BLOCK_MAGIC);
block.push(CACHE_BLOCK_VERSION);
block.push(codec);
block.extend_from_slice(&(value.len() as u64).to_le_bytes());
block.extend_from_slice(&(payload.len() as u64).to_le_bytes());
block.extend_from_slice(&payload);
block.extend_from_slice(payload);
Ok(block)
}

Expand Down Expand Up @@ -362,6 +375,136 @@ fn sync_parent_dir(path: &Path) -> std::io::Result<()> {
Ok(())
}

/// Decoded block bytes, in the form the read that asked for them needs.
///
/// A read that refills hands the bytes to **two** holders: the tier it refills
/// and the caller. Decoding into the handle the tier wants costs one
/// allocation for the tier and one copy for the caller.
///
/// A read that cannot refill -- an SSD-only instance, which never puts
/// anything back into memory -- has one holder, and a handle would have to be
/// copied again to hand back an owned value. That read decodes into the value
/// it is going to return.
///
/// Both of those used to be the same thing: decode into a `Vec`, clone the
/// `Vec` for the tier, and let the tier build an `Arc` from the clone. Three
/// payload-sized allocations to serve two holders, and the extra one grew with
/// the payload.
enum CacheDecodedBlock {
/// Shared with the tier this read refills.
Shared(Arc<[u8]>),
/// Owned by the read, because nothing else wants these bytes.
Owned(Vec<u8>),
}

impl CacheDecodedBlock {
fn decode(block: &[u8], refilling: bool) -> Result<Self, CacheError> {
if refilling {
Ok(Self::Shared(decode_cache_block_shared(block)?))
} else {
Ok(Self::Owned(decode_cache_block(block)?))
}
}

fn len(&self) -> usize {
match self {
Self::Shared(bytes) => bytes.len(),
Self::Owned(bytes) => bytes.len(),
}
}

/// A handle to refill a tier with, or `None` from the form that has none.
///
/// A read that asked for the owning form is a read that said it would not
/// refill, so there is nothing to hand over.
fn to_refill(&self) -> Option<Arc<[u8]>> {
match self {
Self::Shared(bytes) => Some(Arc::clone(bytes)),
Self::Owned(_) => None,
}
}

/// One owned copy, for a read that needs several.
fn to_vec(&self) -> Vec<u8> {
match self {
Self::Shared(bytes) => bytes.to_vec(),
Self::Owned(bytes) => bytes.clone(),
}
}

/// The owned copy, for a read that needs one.
///
/// The owning form moves rather than copies here, which is why the two
/// forms are worth telling apart at all.
fn into_vec(self) -> Vec<u8> {
match self {
Self::Shared(bytes) => bytes.to_vec(),
Self::Owned(bytes) => bytes,
}
}
}

/// Decode a block into a handle the tiers and the caller share.
///
/// Seven read paths wrote `Arc::<[u8]>::from(decode_cache_block(..)?)`, which
/// for an uncompressed block is two payload-sized allocations and two copies:
/// one into the `Vec` the decode owns, one into the `Arc` that `Vec` is thrown
/// away for. An uncompressed payload is already a slice of the block, so the
/// handle can be built straight from it.
///
/// A compressed payload still decompresses into a `Vec` first -- building the
/// handle in place without one would need the unsafe this crate forbids.
fn decode_cache_block_shared(block: &[u8]) -> Result<Arc<[u8]>, CacheError> {
if let Some(payload) = uncompressed_block_payload(block)? {
return Ok(Arc::<[u8]>::from(payload));
}
Ok(Arc::<[u8]>::from(decode_cache_block(block)?))
}

/// The payload of a block that needs no decompressing, or `None` when it does.
///
/// The header checks are the ones [`decode_cache_block`] makes, in the same
/// order, so a block either passes both or fails both.
fn uncompressed_block_payload(block: &[u8]) -> Result<Option<&[u8]>, CacheError> {
if !block.starts_with(CACHE_BLOCK_MAGIC) {
return Ok(Some(block));
}
if block.len() < CACHE_HEADER_LEN {
return Err(CacheError::CorruptBlock("short header".to_string()));
}
let version = block[8];
if version != CACHE_BLOCK_VERSION {
return Err(CacheError::CorruptBlock(format!(
"unsupported version {version}"
)));
}
if block[9] != CACHE_CODEC_RAW {
return Ok(None);
}
let original_len = u64::from_le_bytes(
block[10..18]
.try_into()
.expect("cache block original length slice"),
) as usize;
let payload_len = u64::from_le_bytes(
block[18..26]
.try_into()
.expect("cache block payload length slice"),
) as usize;
if block.len() != CACHE_HEADER_LEN + payload_len {
return Err(CacheError::CorruptBlock(
"payload length mismatch".to_string(),
));
}
let payload = &block[CACHE_HEADER_LEN..];
if payload.len() != original_len {
return Err(CacheError::CorruptBlock(
"original length mismatch".to_string(),
));
}
Ok(Some(payload))
}

fn decode_cache_block(block: &[u8]) -> Result<Vec<u8>, CacheError> {
if !block.starts_with(CACHE_BLOCK_MAGIC) {
return Ok(block.to_vec());
Expand Down Expand Up @@ -2039,10 +2182,6 @@ impl CacheInner {
self.put_ssd_bypass_if_absent(key, value)
}

fn refill_from_ssd(&mut self, key: &CacheKey, value: Vec<u8>) -> bool {
self.refill_from_ssd_shared(key, Arc::<[u8]>::from(value))
}

fn refill_from_ssd_shared(&mut self, key: &CacheKey, value: Arc<[u8]>) -> bool {
if matches!(
self.tiering_policy.data_placement,
Expand Down
Loading
Loading