You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafficserver.apache.org by mo...@apache.org on 2026/09/22 18:00:12 UTC
(trafficserver) branch master updated: Honor the RamCache copy flag in LRU and S3-FIFO (#13380)
This is an automated email from the ASF dual-hosted git repository.
moonchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/trafficserver.git
The following commit(s) were added to refs/heads/master by this push:
new 493bc01d1b Honor the RamCache copy flag in LRU and S3-FIFO (#13380)
493bc01d1b is described below
commit 493bc01d1b8ee412507f2776f0e8a702fb99686d
Author: Phong Nguyen <ph...@gmail.com>
AuthorDate: Tue Sep 22 11:00:04 2026 -0700
Honor the RamCache copy flag in LRU and S3-FIFO (#13380)
CacheVC requests copy semantics (put copy=true) whenever RAM cache
compression is configured, because it unmarshals HTTP headers in place
in its own buffer after the put and again on every RAM hit. LRU and
S3-FIFO ignored the flag and shared buffers with the caller in both
directions, so with proxy.config.cache.ram_cache.compress enabled and
algorithm=1 (the default) or 2, the cached bytes were mutated after
insertion and every subsequent HTTP RAM hit re-unmarshalled an
already-unmarshalled header (assert in debug builds, header unmarshal
failure in release). Copy entries now copy on put, copy on get, and a
copy=true put refreshes a resident entry that may still be sharing a
caller's buffer, matching CLFUS.
Co-Authored-By: Claude Fable 5 <no...@anthropic.com>
---
src/iocore/cache/CMakeLists.txt | 1 +
src/iocore/cache/P_RamCache.h | 29 ++
src/iocore/cache/RamCacheCLFUS.cc | 31 +-
src/iocore/cache/RamCacheLRU.cc | 49 +-
src/iocore/cache/RamCacheS3FIFO.cc | 50 +-
src/iocore/cache/unit_tests/test_RamCacheCopy.cc | 562 +++++++++++++++++++++++
6 files changed, 698 insertions(+), 24 deletions(-)
diff --git a/src/iocore/cache/CMakeLists.txt b/src/iocore/cache/CMakeLists.txt
index e2e4c263ce..0b5f66fc57 100644
--- a/src/iocore/cache/CMakeLists.txt
+++ b/src/iocore/cache/CMakeLists.txt
@@ -103,6 +103,7 @@ if(BUILD_TESTING)
add_cache_test(CacheStripe unit_tests/test_Stripe.cc)
add_cache_test(CacheAggregateWriteBuffer unit_tests/test_AggregateWriteBuffer.cc)
add_cache_test(RamCacheCLFUS unit_tests/test_RamCacheCLFUS.cc)
+ add_cache_test(RamCacheCopy unit_tests/test_RamCacheCopy.cc)
add_cache_test(RamCacheCompressEntries unit_tests/test_RamCacheCompressEntries.cc)
add_cache_test(RamCacheSeenFilter unit_tests/test_RamCacheSeenFilter.cc)
# Only the shutdown test attaches a live segment; the rest need no shm syscall.
diff --git a/src/iocore/cache/P_RamCache.h b/src/iocore/cache/P_RamCache.h
index 1afe103751..1af9bce192 100644
--- a/src/iocore/cache/P_RamCache.h
+++ b/src/iocore/cache/P_RamCache.h
@@ -28,6 +28,8 @@
#include "iocore/eventsystem/IOBuffer.h"
#include "tscore/CryptoHash.h"
+#include <cstring>
+
class StripeSM;
class RamCache
@@ -41,6 +43,33 @@ public:
virtual void init(int64_t max_bytes, StripeSM *stripe) = 0;
virtual ~RamCache(){};
+
+protected:
+ // put(copy = true) is the caller's promise that it may mutate its buffer
+ // after the put, so buffers are never shared with callers in either
+ // direction for such entries: implementations store a private copy on put
+ // (copy_data_in) and hand out a private copy on get (copy_data_out).
+
+ // Exact-size private copy of the caller's buffer; block_size() on the
+ // result recovers len.
+ static Ptr<IOBufferData>
+ copy_data_in(IOBufferData *data, uint32_t len)
+ {
+ Ptr<IOBufferData> d = make_ptr(new_IOBufferData(BUFFER_SIZE_INDEX_FOR_XMALLOC_SIZE(len), DEFAULT_ALLOC));
+
+ memcpy(d->data(), data->data(), len);
+ return d;
+ }
+
+ // Private copy of a stored buffer to hand to the caller.
+ static Ptr<IOBufferData>
+ copy_data_out(IOBufferData *data, uint32_t len)
+ {
+ Ptr<IOBufferData> d = make_ptr(new_IOBufferData(iobuffer_size_to_index(len, MAX_BUFFER_SIZE_INDEX), MEMALIGNED));
+
+ memcpy(d->data(), data->data(), len);
+ return d;
+ }
};
RamCache *new_RamCacheLRU();
diff --git a/src/iocore/cache/RamCacheCLFUS.cc b/src/iocore/cache/RamCacheCLFUS.cc
index 82146e2753..90254841b7 100644
--- a/src/iocore/cache/RamCacheCLFUS.cc
+++ b/src/iocore/cache/RamCacheCLFUS.cc
@@ -426,12 +426,11 @@ RamCacheCLFUS::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxkey
}
(*ret_data) = data;
} else {
- IOBufferData *data = e->data.get();
if (e->flag_bits.copy) {
- data = new_IOBufferData(iobuffer_size_to_index(e->len, MAX_BUFFER_SIZE_INDEX), MEMALIGNED);
- ::memcpy(data->data(), e->data->data(), e->len);
+ (*ret_data) = copy_data_out(e->data.get(), e->len);
+ } else {
+ (*ret_data) = e->data;
}
- (*ret_data) = data;
}
ts::Metrics::Counter::increment(cache_rsb.ram_cache_hits);
ts::Metrics::Counter::increment(stripe->cache_vol->vol_rsb.ram_cache_hits);
@@ -811,6 +810,17 @@ RamCacheCLFUS::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy,
this->_move_compressed(e);
this->_lru[e->flag_bits.lru].remove(e);
this->_lru[e->flag_bits.lru].enqueue(e);
+ if (copy && e->flag_bits.copy && !e->flag_bits.compressed) {
+ // Already an uncompressed private copy of this object: get() never
+ // exposes that buffer, so re-copying it (two requests that both missed
+ // and both read the object from disk) changes nothing. Everything
+ // below is a no-op for such an entry apart from the allocation and
+ // memcpy -- size is already len, so delta is 0 -- except when the
+ // entry is compressed, where the swap is what decompresses it and
+ // skipping it would leave compressed bytes behind a cleared flag.
+ DDbg(dbg_ctl_ram_cache, "put %X %" PRId64 " size %d HIT (private, unchanged)", key->slice32(3), auxkey, e->size);
+ return 1;
+ }
int64_t delta = (static_cast<int64_t>(size)) - static_cast<int64_t>(e->size);
this->_bytes += delta;
ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, delta);
@@ -819,12 +829,10 @@ RamCacheCLFUS::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy,
e->size = size;
e->data = data;
} else {
- char *b = static_cast<char *>(ats_malloc(len));
- memcpy(b, data->data(), len);
- e->data = new_xmalloc_IOBufferData(b, len);
- e->data->_mem_type = DEFAULT_ALLOC;
- e->size = size;
+ e->data = copy_data_in(data, len);
+ e->size = size;
}
+ e->len = len; // get() and the compressor read e->len bytes out of e->data
check_accounting(this);
e->flag_bits.copy = copy;
e->flag_bits.compressed = 0;
@@ -930,10 +938,7 @@ Linsert:
if (!copy) {
e->data = data;
} else {
- char *b = static_cast<char *>(ats_malloc(len));
- memcpy(b, data->data(), len);
- e->data = new_xmalloc_IOBufferData(b, len);
- e->data->_mem_type = DEFAULT_ALLOC;
+ e->data = copy_data_in(data, len);
}
e->flag_bits.copy = copy;
this->_bytes += size + entry_overhead;
diff --git a/src/iocore/cache/RamCacheLRU.cc b/src/iocore/cache/RamCacheLRU.cc
index 355df2ae5f..1cb1363626 100644
--- a/src/iocore/cache/RamCacheLRU.cc
+++ b/src/iocore/cache/RamCacheLRU.cc
@@ -33,6 +33,8 @@
struct RamCacheLRUEntry {
CryptoHash key;
uint64_t auxkey;
+ uint32_t len; // data length, so get() does not depend on copy_data_in allocating exactly len
+ bool copy; // copy-in-copy-out: buffers are never shared with callers
LINK(RamCacheLRUEntry, lru_link);
LINK(RamCacheLRUEntry, hash_link);
Ptr<IOBufferData> data;
@@ -147,7 +149,11 @@ RamCacheLRU::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxkey)
if (e->key == *key && e->auxkey == auxkey) {
lru.remove(e);
lru.enqueue(e);
- (*ret_data) = e->data;
+ if (e->copy) {
+ (*ret_data) = copy_data_out(e->data.get(), e->len);
+ } else {
+ (*ret_data) = e->data;
+ }
DDbg(dbg_ctl_ram_cache, "get %X %" PRIu64 " HIT", key->slice32(3), auxkey);
ts::Metrics::Counter::increment(cache_rsb.ram_cache_hits);
ts::Metrics::Counter::increment(stripe->cache_vol->vol_rsb.ram_cache_hits);
@@ -181,9 +187,8 @@ RamCacheLRU::remove(RamCacheLRUEntry *e)
return ret;
}
-// ignore 'copy' since we don't touch the data
int
-RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t len, bool, uint64_t auxkey)
+RamCacheLRU::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy, uint64_t auxkey)
{
if (!max_bytes) {
return 0;
@@ -196,6 +201,30 @@ RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t
if (e->auxkey == auxkey) {
lru.remove(e);
lru.enqueue(e);
+ if (copy && !e->copy) {
+ // The entry may still be sharing a caller's buffer from a put made
+ // while copy semantics were not requested; refresh it with a
+ // private copy before the caller mutates its buffer. Once it holds
+ // a private copy there is nothing to refresh: get() never exposes
+ // that buffer, so a further copy=true put (two requests that both
+ // missed and both read the object from disk) would only allocate
+ // and copy the object again to the same effect.
+ //
+ // A refresh only ever gives bytes back, so unlike an insert it needs
+ // no eviction pass: the private copy is an exact-size allocation
+ // while a shared buffer was charged its rounded block_size() (always
+ // >= len), and a re-put under the same key and auxkey names the same
+ // on-disk doc, so len itself does not grow.
+ Ptr<IOBufferData> d = copy_data_in(data, len);
+ int64_t delta = d->block_size() - e->data->block_size();
+
+ e->data = d;
+ e->len = len;
+ e->copy = true;
+ bytes += delta;
+ ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, delta);
+ ts::Metrics::Gauge::increment(stripe->cache_vol->vol_rsb.ram_cache_bytes, delta);
+ }
return 1;
} else { // discard when aux keys conflict
e = remove(e);
@@ -233,13 +262,19 @@ RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t
e = THREAD_ALLOC(ramCacheLRUEntryAllocator, this_ethread());
e->key = *key;
e->auxkey = auxkey;
- e->data = data;
+ e->len = len;
+ e->copy = copy;
+ if (copy) {
+ e->data = copy_data_in(data, len);
+ } else {
+ e->data = data;
+ }
bucket[i].push(e);
lru.enqueue(e);
- bytes += ENTRY_OVERHEAD + data->block_size();
+ bytes += ENTRY_OVERHEAD + e->data->block_size();
objects++;
- ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, ENTRY_OVERHEAD + data->block_size());
- ts::Metrics::Gauge::increment(stripe->cache_vol->vol_rsb.ram_cache_bytes, ENTRY_OVERHEAD + data->block_size());
+ ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, ENTRY_OVERHEAD + e->data->block_size());
+ ts::Metrics::Gauge::increment(stripe->cache_vol->vol_rsb.ram_cache_bytes, ENTRY_OVERHEAD + e->data->block_size());
while (bytes > max_bytes) {
RamCacheLRUEntry *ee = lru.dequeue();
if (ee) {
diff --git a/src/iocore/cache/RamCacheS3FIFO.cc b/src/iocore/cache/RamCacheS3FIFO.cc
index 7471165107..993653e45d 100644
--- a/src/iocore/cache/RamCacheS3FIFO.cc
+++ b/src/iocore/cache/RamCacheS3FIFO.cc
@@ -61,6 +61,7 @@ struct RamCacheS3FIFOEntry {
uint32_t size; // object bytes; resident entries account size + ENTRY_OVERHEAD against the budget
uint8_t seg; // SEG_SMALL / SEG_MAIN / SEG_GHOST
uint8_t freq; // 0..FREQ_MAX
+ bool copy; // copy-in-copy-out: buffers are never shared with callers
LINK(RamCacheS3FIFOEntry, lru_link);
LINK(RamCacheS3FIFOEntry, hash_link);
Ptr<IOBufferData> data; // null for ghost entries
@@ -328,7 +329,13 @@ RamCacheS3FIFO::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxke
if (e->freq < FREQ_MAX) {
e->freq++;
}
- (*ret_data) = e->data;
+ if (e->copy) {
+ // For copy entries the stored buffer is an exact-size allocation, so
+ // e->size is the data length.
+ (*ret_data) = copy_data_out(e->data.get(), e->size);
+ } else {
+ (*ret_data) = e->data;
+ }
ts::Metrics::Counter::increment(cache_rsb.ram_cache_hits);
ts::Metrics::Counter::increment(_stripe->cache_vol->vol_rsb.ram_cache_hits);
return 1;
@@ -339,12 +346,12 @@ RamCacheS3FIFO::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxke
}
int
-RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t len, bool, uint64_t auxkey)
+RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy, uint64_t auxkey)
{
if (!_max_bytes) {
return 0;
}
- uint32_t size = data->block_size();
+ uint32_t size = copy ? len : data->block_size();
uint32_t i = key->slice32(3) % _nbuckets;
// Walk the hash chain. A resident hit just counts the reference; a ghost hit is admitted fresh to
@@ -359,6 +366,36 @@ RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32
if (e->freq < FREQ_MAX) {
e->freq++;
}
+ if (copy && !e->copy) {
+ // The entry may still be sharing a caller's buffer from a put
+ // made while copy semantics were not requested; refresh it with a
+ // private copy before the caller mutates its buffer. Once it holds
+ // a private copy there is nothing to refresh: get() never exposes
+ // that buffer, so a further copy=true put (two requests that both
+ // missed and both read the object from disk) would only allocate
+ // and copy the object again to the same effect.
+ //
+ // A refresh only ever gives bytes back, so unlike an insert it
+ // needs no eviction pass: the private copy is charged its exact
+ // length while a shared buffer was charged the rounded
+ // block_size() (always >= len), and a re-put under the same key
+ // and auxkey names the same on-disk doc, so len itself does not
+ // grow. That matters here because the insert path keeps the budget
+ // by declining an oversized object outright, which a refresh of an
+ // already-resident entry could not do.
+ int64_t delta = static_cast<int64_t>(len) - e->size;
+
+ e->data = copy_data_in(data, len);
+ e->copy = true;
+ e->size = len;
+ if (e->seg == SEG_SMALL) {
+ _s_bytes += delta;
+ } else {
+ _m_bytes += delta;
+ }
+ ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, delta);
+ ts::Metrics::Gauge::increment(_stripe->cache_vol->vol_rsb.ram_cache_bytes, delta);
+ }
return 1;
}
RamCacheS3FIFOEntry *next = e->hash_link.next; // ghost hit: admit a fresh copy to main
@@ -396,7 +433,12 @@ RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32
ne->size = size;
ne->freq = 0;
ne->seg = ghost_hit ? SEG_MAIN : SEG_SMALL;
- ne->data = data;
+ ne->copy = copy;
+ if (copy) {
+ ne->data = copy_data_in(data, len);
+ } else {
+ ne->data = data;
+ }
_bucket[i].push(ne);
_seg[ne->seg].enqueue(ne);
if (ghost_hit) {
diff --git a/src/iocore/cache/unit_tests/test_RamCacheCopy.cc b/src/iocore/cache/unit_tests/test_RamCacheCopy.cc
new file mode 100644
index 0000000000..dc457f4736
--- /dev/null
+++ b/src/iocore/cache/unit_tests/test_RamCacheCopy.cc
@@ -0,0 +1,562 @@
+/** @file
+
+ Catch-based unit tests for the RamCache `copy` contract across all policies.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+// put(..., copy = true) is the caller's promise that it may mutate its buffer
+// after the put (CacheVC unmarshals HTTP headers in place after inserting into
+// the RAM cache when compression is configured, see CacheVC.cc). The cache
+// must therefore never share buffers with the caller in either direction for
+// copy entries: it copies on put and copies on get. These tests pin that
+// contract for every RamCache implementation.
+
+#include "main.h"
+#include "test_doubles.h"
+
+#include "../P_CacheInternal.h"
+#include "../P_RamCache.h"
+
+#include "tscore/ink_config.h"
+
+#include <cstdint>
+#include <cstring>
+#include <vector>
+
+// Required by main.h
+int cache_vols = 1;
+bool reuse_existing_cache = false;
+
+namespace
+{
+
+// Deliberately not a power of two. Every policy charges block_size() for a
+// buffer it shares with the caller and len for a private copy, so a payload
+// that lands exactly on a size index makes all of the copy-related size
+// arithmetic a no-op (delta == 0) and leaves it untested.
+constexpr std::size_t PAYLOAD_LEN = 5000;
+
+struct PolicyCase {
+ RamCache *(*factory)();
+ const char *name;
+ // Whether size() is derived from the same running counters the refresh
+ // adjusts, so it can be checked against the gauge. LRU recomputes size() by
+ // walking its entry list with a different per-entry formula than the counter
+ // that drives eviction, and CLFUS counts per-entry overhead in size() but
+ // not in the gauge; for both, size() cannot witness a bad refresh delta.
+ bool size_tracks_gauge;
+};
+
+const PolicyCase policy_cases[] = {
+ {new_RamCacheLRU, "LRU", false},
+ {new_RamCacheCLFUS, "CLFUS", false},
+ {new_RamCacheS3FIFO, "S3FIFO", true },
+};
+
+// The RamCache get/put paths touch only these metrics and the stripe mutex.
+void
+wire_stripe(StripeSM &stripe, CacheVol &cache_vol)
+{
+ stripe.cache_vol = &cache_vol;
+
+ cache_rsb.ram_cache_bytes = ts::Metrics::Gauge::createPtr("unit_test.copy.ram_cache.bytes");
+ cache_rsb.ram_cache_hits = ts::Metrics::Counter::createPtr("unit_test.copy.ram_cache.hits");
+ cache_rsb.ram_cache_misses = ts::Metrics::Counter::createPtr("unit_test.copy.ram_cache.misses");
+ cache_vol.vol_rsb.ram_cache_bytes = ts::Metrics::Gauge::createPtr("unit_test.copy.vol.ram_cache.bytes");
+ cache_vol.vol_rsb.ram_cache_hits = ts::Metrics::Counter::createPtr("unit_test.copy.vol.ram_cache.hits");
+ cache_vol.vol_rsb.ram_cache_misses = ts::Metrics::Counter::createPtr("unit_test.copy.vol.ram_cache.misses");
+}
+
+std::vector<char>
+pattern_bytes(std::size_t len, char base)
+{
+ std::vector<char> bytes(len);
+
+ for (std::size_t i = 0; i < len; i++) {
+ bytes[i] = static_cast<char>(base + (i % 26));
+ }
+ return bytes;
+}
+
+Ptr<IOBufferData>
+make_buffer(const std::vector<char> &bytes)
+{
+ int64_t idx = iobuffer_size_to_index(bytes.size(), MAX_BUFFER_SIZE_INDEX);
+ Ptr<IOBufferData> data{make_ptr(new_IOBufferData(idx, MEMALIGNED))};
+
+ std::memcpy(data->data(), bytes.data(), bytes.size());
+ return data;
+}
+
+// What make_buffer()'s rounded-up allocation costs the cache: the amount a
+// copy=false put of PAYLOAD_LEN bytes is charged, read off the same accessor
+// the policies charge.
+int64_t
+payload_block_len()
+{
+ return make_buffer(std::vector<char>(PAYLOAD_LEN))->block_size();
+}
+
+// Bytes a copy=true put gives back relative to a copy=false put of the same
+// object, because the private copy is an exact-size allocation.
+int64_t
+copy_savings()
+{
+ return payload_block_len() - static_cast<int64_t>(PAYLOAD_LEN);
+}
+
+CryptoHash
+fresh_key()
+{
+ static uint64_t salt = 0;
+
+ ++salt;
+
+ // The policies bucket on slice32(3), the high half of u64[1]; vary it so
+ // entries spread across hash buckets.
+ CryptoHash key;
+
+ key.u64[0] = 0xc0ffee00 + salt;
+ key.u64[1] = (uint64_t{0xdeadbeef} + salt) << 32 | 0x5eed;
+ return key;
+}
+
+RamCache *
+make_cache(RamCache *(*factory)(), StripeSM &stripe, int64_t max_bytes = 1 << 20)
+{
+ // No compression: CLFUS must not schedule its background compressor (which
+ // would retain a pointer to this cache), and the seen filter would
+ // otherwise reject first-time puts.
+ cache_config_ram_cache_compress = 0;
+ cache_config_ram_cache_use_seen_filter = 0;
+
+ // The policies have no destructors (entries are pool-allocated and only
+ // released on eviction), so destroying a cache object strands its entries
+ // for leak checkers. Keep every cache reachable for the life of the
+ // process instead.
+ static std::vector<RamCache *> &all_caches = *new std::vector<RamCache *>;
+ RamCache *rc = factory();
+
+ all_caches.push_back(rc);
+ rc->init(max_bytes, &stripe);
+ return rc;
+}
+
+} // namespace
+
+TEST_CASE("RamCache copy=true entries are immune to caller-side mutation after put", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ const PolicyCase pc = GENERATE(from_range(std::begin(policy_cases), std::end(policy_cases)));
+ INFO("policy: " << pc.name);
+
+ auto rc = make_cache(pc.factory, stripe);
+ auto payload = pattern_bytes(PAYLOAD_LEN, 'A');
+ auto buf = make_buffer(payload);
+ auto key = fresh_key();
+
+ REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1);
+
+ // The caller mutates its buffer after the put, exactly as CacheVC does when
+ // it unmarshals HTTP headers in place.
+ std::memset(buf->data(), 0x5a, payload.size());
+
+ Ptr<IOBufferData> got;
+
+ REQUIRE(rc->get(&key, &got) >= 1);
+ REQUIRE(got.get() != nullptr);
+ CHECK(std::memcmp(got->data(), payload.data(), payload.size()) == 0);
+}
+
+TEST_CASE("RamCache copy=true entries are immune to caller-side mutation after get", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ const PolicyCase pc = GENERATE(from_range(std::begin(policy_cases), std::end(policy_cases)));
+ INFO("policy: " << pc.name);
+
+ auto rc = make_cache(pc.factory, stripe);
+ auto payload = pattern_bytes(PAYLOAD_LEN, 'A');
+ auto buf = make_buffer(payload);
+ auto key = fresh_key();
+
+ REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1);
+
+ Ptr<IOBufferData> first;
+
+ REQUIRE(rc->get(&key, &first) >= 1);
+ REQUIRE(first.get() != nullptr);
+ // The caller mutates the buffer it was handed, as it does when unmarshalling
+ // a RAM-cache hit in place.
+ std::memset(first->data(), 0x5a, payload.size());
+
+ Ptr<IOBufferData> second;
+
+ REQUIRE(rc->get(&key, &second) >= 1);
+ REQUIRE(second.get() != nullptr);
+ CHECK(std::memcmp(second->data(), payload.data(), payload.size()) == 0);
+}
+
+TEST_CASE("RamCache resident entries are refreshed by a copy=true put", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ const PolicyCase pc = GENERATE(from_range(std::begin(policy_cases), std::end(policy_cases)));
+ INFO("policy: " << pc.name);
+
+ auto rc = make_cache(pc.factory, stripe);
+ auto payload = pattern_bytes(PAYLOAD_LEN, 'A');
+ auto buf = make_buffer(payload);
+ auto key = fresh_key();
+
+ // Entry first stored with copy=false (e.g. compression was disabled), then
+ // the same object is re-put with copy=true after a config change. The cache
+ // must own a private copy from that point on.
+ REQUIRE(rc->put(&key, buf.get(), payload.size(), false) == 1);
+ REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1);
+
+ std::memset(buf->data(), 0x5a, payload.size());
+
+ Ptr<IOBufferData> got;
+
+ REQUIRE(rc->get(&key, &got) >= 1);
+ REQUIRE(got.get() != nullptr);
+ CHECK(std::memcmp(got->data(), payload.data(), payload.size()) == 0);
+}
+
+TEST_CASE("RamCache copy=false entries still share the caller's buffer", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ const PolicyCase pc = GENERATE(from_range(std::begin(policy_cases), std::end(policy_cases)));
+ INFO("policy: " << pc.name);
+
+ auto rc = make_cache(pc.factory, stripe);
+ auto payload = pattern_bytes(PAYLOAD_LEN, 'A');
+ auto buf = make_buffer(payload);
+ auto key = fresh_key();
+
+ REQUIRE(rc->put(&key, buf.get(), payload.size(), false) == 1);
+
+ Ptr<IOBufferData> got;
+
+ REQUIRE(rc->get(&key, &got) >= 1);
+ REQUIRE(got.get() != nullptr);
+ // Zero-copy is the point of copy=false: the cache hands back the same
+ // buffer it was given.
+ CHECK(got->data() == buf->data());
+}
+
+TEST_CASE("RamCache byte accounting survives the copy=true resident refresh", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ const PolicyCase pc = GENERATE(from_range(std::begin(policy_cases), std::end(policy_cases)));
+ INFO("policy: " << pc.name);
+
+ // The refresh is the only place the fix does arithmetic rather than just
+ // swapping a buffer, so pin the gauge across it. Caches from earlier test
+ // cases are kept alive but idle, so zeroing the gauges here leaves this
+ // cache as the only writer and the values below can be read absolutely.
+ ts::Metrics::Gauge::store(cache_rsb.ram_cache_bytes, 0);
+ ts::Metrics::Gauge::store(cache_vol.vol_rsb.ram_cache_bytes, 0);
+
+ constexpr int64_t cache_bytes = 128 * 1024;
+ auto rc = make_cache(pc.factory, stripe, cache_bytes);
+ auto payload = pattern_bytes(PAYLOAD_LEN, 'A');
+
+ // Small enough to fit without evicting, so the numbers below are only the
+ // refresh's doing.
+ constexpr int n_objects = 8;
+ std::vector<CryptoHash> keys;
+ std::vector<Ptr<IOBufferData>> bufs;
+
+ for (int i = 0; i < n_objects; i++) {
+ keys.push_back(fresh_key());
+ bufs.push_back(make_buffer(payload));
+ REQUIRE(rc->put(&keys[i], bufs[i].get(), payload.size(), false) == 1);
+ }
+
+ const int64_t after_puts = ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes);
+
+ CHECK(after_puts > 0);
+ CHECK(rc->size() > 0);
+ if (pc.size_tracks_gauge) {
+ // Nothing has been evicted, so there are no ghosts and size() is exactly
+ // the resident accounting the gauge reports.
+ CHECK(rc->size() == after_puts);
+ }
+
+ // Re-put each object with copy=true. Every policy charges block_size() for a
+ // shared buffer and len for its own exact-size copy, so each refresh must
+ // hand back exactly the size-index rounding and nothing else.
+ for (int i = 0; i < n_objects; i++) {
+ REQUIRE(rc->put(&keys[i], bufs[i].get(), payload.size(), true) == 1);
+ }
+
+ const int64_t after_refresh = ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes);
+
+ CHECK(after_refresh == after_puts - n_objects * copy_savings());
+ // The per-volume gauge is updated alongside the global one on every path.
+ CHECK(ts::Metrics::Gauge::load(cache_vol.vol_rsb.ram_cache_bytes) == after_refresh);
+ CHECK(rc->size() > 0);
+ if (pc.size_tracks_gauge) {
+ // The gauge alone cannot see a refresh that forgets to credit the counter
+ // eviction runs off, because the two are updated independently.
+ CHECK(rc->size() == after_refresh);
+ }
+
+ // Overflow the cache so the refreshed entries are evicted: eviction has to
+ // subtract what the refresh left behind, or the gauge drifts negative.
+ constexpr int flood = 64;
+
+ for (int i = 0; i < flood; i++) {
+ CryptoHash k = fresh_key();
+ auto b = make_buffer(payload);
+
+ // A put may legitimately be declined once the cache is full (CLFUS weighs
+ // the new object against its victims), which is not what is under test.
+ rc->put(&k, b.get(), payload.size(), false);
+ CHECK(ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes) >= 0);
+ CHECK(rc->size() >= 0);
+ }
+
+ CHECK(ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes) >= 0);
+ CHECK(ts::Metrics::Gauge::load(cache_vol.vol_rsb.ram_cache_bytes) >= 0);
+ CHECK(rc->size() >= 0);
+}
+
+TEST_CASE("RamCacheS3FIFO honors copy on a ghost readmit", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ ts::Metrics::Gauge::store(cache_rsb.ram_cache_bytes, 0);
+ ts::Metrics::Gauge::store(cache_vol.vol_rsb.ram_cache_bytes, 0);
+
+ // A ghost readmit is the one insert that lands in the main queue, so it is
+ // the only way `copy` accounting reaches _m_bytes. Charging len instead of
+ // block_size() there is otherwise untested.
+ constexpr int64_t cache_bytes = 128 * 1024;
+ auto rc = make_cache(new_RamCacheS3FIFO, stripe, cache_bytes);
+ auto payload = pattern_bytes(PAYLOAD_LEN, 'A');
+
+ auto key = fresh_key();
+ auto buf = make_buffer(payload);
+
+ // Stored shared first, so the readmit below also crosses from a
+ // block_size()-charged entry to a len-charged one.
+ REQUIRE(rc->put(&key, buf.get(), payload.size(), false) == 1);
+
+ // Push the object out of the small queue without ever referencing it, so it
+ // is demoted to the ghost rather than promoted. The cache holds roughly
+ // cache_bytes / (ENTRY_OVERHEAD + block_size) objects, and the default ghost
+ // bounds (ghost_size_percent 90, ghost_mem_percent 25) are far from binding
+ // at this flood size, so the key is still a ghost afterwards.
+ constexpr int flood = 20;
+
+ for (int i = 0; i < flood; i++) {
+ CryptoHash k = fresh_key();
+ auto b = make_buffer(payload);
+
+ REQUIRE(rc->put(&k, b.get(), payload.size(), false) == 1);
+ }
+
+ Ptr<IOBufferData> evicted;
+
+ // A ghost entry holds a key but no data, so it reads as a miss.
+ REQUIRE(rc->get(&key, &evicted) == 0);
+
+ // Ghost hit: the stale entry is removed and a fresh one is admitted straight
+ // to the main queue, inheriting the copy handling from the shared insert
+ // site.
+ REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1);
+ CHECK(ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes) >= 0);
+ CHECK(ts::Metrics::Gauge::load(cache_vol.vol_rsb.ram_cache_bytes) == ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes));
+ CHECK(rc->size() >= 0);
+ CHECK(rc->size() <= cache_bytes);
+
+ // The readmitted entry must own its data like any other copy entry.
+ std::memset(buf->data(), 0x5a, payload.size());
+
+ Ptr<IOBufferData> got;
+
+ REQUIRE(rc->get(&key, &got) >= 1);
+ REQUIRE(got.get() != nullptr);
+ CHECK(std::memcmp(got->data(), payload.data(), payload.size()) == 0);
+
+ // Evidence that the readmit landed in the main queue: small-queue pressure
+ // does not touch it. An entry sitting in the small queue with a reuse count
+ // below promote_threshold would have been demoted to the ghost by now.
+ for (int i = 0; i < flood; i++) {
+ CryptoHash k = fresh_key();
+ auto b = make_buffer(payload);
+
+ REQUIRE(rc->put(&k, b.get(), payload.size(), false) == 1);
+ CHECK(ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes) >= 0);
+ }
+
+ Ptr<IOBufferData> survived;
+
+ CHECK(rc->get(&key, &survived) >= 1);
+ CHECK(ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes) >= 0);
+ CHECK(rc->size() >= 0);
+}
+
+TEST_CASE("RamCache refresh credits the counter eviction runs off", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ const PolicyCase pc = GENERATE(from_range(std::begin(policy_cases), std::end(policy_cases)));
+ INFO("policy: " << pc.name);
+
+ // The gauge is only reported; the refresh also has to credit the internal
+ // counter that admission and eviction are driven by, and for LRU that
+ // counter is not observable through size(). Check it by behavior instead:
+ // fill the cache to just under its budget with shared buffers, refresh every
+ // entry to a private copy, then insert more objects than there was room for
+ // beforehand. They only fit if the refresh really did give the bytes back,
+ // and if it did, nothing resident is evicted to make space.
+ constexpr int64_t cache_bytes = 128 * 1024;
+ auto rc = make_cache(pc.factory, stripe, cache_bytes);
+ auto payload = pattern_bytes(PAYLOAD_LEN, 'A');
+
+ // n_objects shared copies just fit; the n_extra that follow fit only out of
+ // what the refreshes free (copy_savings() per object).
+ constexpr int n_objects = 15;
+ constexpr int n_extra = 5;
+ std::vector<CryptoHash> keys;
+ std::vector<Ptr<IOBufferData>> bufs;
+
+ for (int i = 0; i < n_objects; i++) {
+ keys.push_back(fresh_key());
+ bufs.push_back(make_buffer(payload));
+ REQUIRE(rc->put(&keys[i], bufs[i].get(), payload.size(), false) == 1);
+ }
+
+ REQUIRE(n_objects * copy_savings() > n_extra * payload_block_len());
+
+ for (int i = 0; i < n_objects; i++) {
+ REQUIRE(rc->put(&keys[i], bufs[i].get(), payload.size(), true) == 1);
+ }
+
+ for (int i = 0; i < n_extra; i++) {
+ CryptoHash k = fresh_key();
+ auto b = make_buffer(payload);
+
+ REQUIRE(rc->put(&k, b.get(), payload.size(), false) == 1);
+ }
+
+ // Every refreshed entry is still resident. Had the refresh left the counter
+ // alone the cache would still believe it was full, and the first of the
+ // extra inserts would have evicted the oldest of these.
+ for (int i = 0; i < n_objects; i++) {
+ Ptr<IOBufferData> got;
+
+ INFO("object " << i);
+ CHECK(rc->get(&keys[i], &got) >= 1);
+ }
+}
+
+TEST_CASE("RamCache copy=true put onto a private entry leaves it alone", "[cache][ramcache][copy]")
+{
+ CacheDisk disk;
+ init_disk(disk);
+ StripeSM stripe{&disk, 10, 0};
+ CacheVol cache_vol;
+ wire_stripe(stripe, cache_vol);
+
+ const PolicyCase pc = GENERATE(from_range(std::begin(policy_cases), std::end(policy_cases)));
+ INFO("policy: " << pc.name);
+
+ // Two requests that both miss and both read the object from disk will both
+ // put it with copy=true; whichever lands second finds a private copy already
+ // resident. Re-copying it is pure cost -- get() never exposes that buffer --
+ // so the refresh is reserved for the shared-to-private transition.
+ //
+ // The cache cannot tell the caller's buffers apart, and the gauge does not
+ // move either way (a private copy is already charged len), so the only way
+ // to see which happened is to re-put different bytes under the same key and
+ // auxkey and read back which ones the cache kept. Real callers never do that;
+ // the same key and auxkey name one on-disk doc.
+ auto rc = make_cache(pc.factory, stripe);
+ auto first = pattern_bytes(PAYLOAD_LEN, 'A');
+ auto second = pattern_bytes(PAYLOAD_LEN, 'a');
+ auto buf1 = make_buffer(first);
+ auto buf2 = make_buffer(second);
+ auto key = fresh_key();
+
+ REQUIRE(rc->put(&key, buf1.get(), first.size(), true) == 1);
+
+ const int64_t before = ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes);
+
+ REQUIRE(rc->put(&key, buf2.get(), second.size(), true) == 1);
+ CHECK(ts::Metrics::Gauge::load(cache_rsb.ram_cache_bytes) == before);
+
+ Ptr<IOBufferData> got;
+
+ REQUIRE(rc->get(&key, &got) >= 1);
+ REQUIRE(got.get() != nullptr);
+ CHECK(std::memcmp(got->data(), first.data(), first.size()) == 0);
+
+ // A shared entry, by contrast, must still be refreshed to a private copy on
+ // the first copy=true put -- the guard is on the entry's state, not on how
+ // many puts it has seen.
+ auto shared = pattern_bytes(PAYLOAD_LEN, 'S');
+ auto shared_buf = make_buffer(shared);
+ auto shared_key = fresh_key();
+
+ REQUIRE(rc->put(&shared_key, shared_buf.get(), shared.size(), false) == 1);
+ REQUIRE(rc->put(&shared_key, shared_buf.get(), shared.size(), true) == 1);
+ std::memset(shared_buf->data(), 0x5a, shared.size());
+
+ Ptr<IOBufferData> got_shared;
+
+ REQUIRE(rc->get(&shared_key, &got_shared) >= 1);
+ CHECK(std::memcmp(got_shared->data(), shared.data(), shared.size()) == 0);
+}