diff --git a/src/iocore/cache/CMakeLists.txt b/src/iocore/cache/CMakeLists.txt index e2e4c263ce2..0b5f66fc57e 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 1afe1037517..1af9bce1927 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 + class StripeSM; class RamCache @@ -41,6 +43,33 @@ class RamCache 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 + copy_data_in(IOBufferData *data, uint32_t len) + { + Ptr 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 + copy_data_out(IOBufferData *data, uint32_t len) + { + Ptr 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 82146e2753e..90254841b77 100644 --- a/src/iocore/cache/RamCacheCLFUS.cc +++ b/src/iocore/cache/RamCacheCLFUS.cc @@ -426,12 +426,11 @@ RamCacheCLFUS::get(CryptoHash *key, Ptr *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(size)) - static_cast(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(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 @@ RamCacheCLFUS::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy, if (!copy) { e->data = data; } else { - char *b = static_cast(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 355df2ae5f7..1cb1363626e 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 data; @@ -147,7 +149,11 @@ RamCacheLRU::get(CryptoHash *key, Ptr *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 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 7471165107b..993653e45dd 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 data; // null for ghost entries @@ -328,7 +329,13 @@ RamCacheS3FIFO::get(CryptoHash *key, Ptr *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 *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(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 00000000000..dc457f4736f --- /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 +#include +#include + +// 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 +pattern_bytes(std::size_t len, char base) +{ + std::vector bytes(len); + + for (std::size_t i = 0; i < len; i++) { + bytes[i] = static_cast(base + (i % 26)); + } + return bytes; +} + +Ptr +make_buffer(const std::vector &bytes) +{ + int64_t idx = iobuffer_size_to_index(bytes.size(), MAX_BUFFER_SIZE_INDEX); + Ptr 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(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(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 &all_caches = *new std::vector; + 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 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 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 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 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 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 keys; + std::vector> 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 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 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 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 keys; + std::vector> 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 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 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 got_shared; + + REQUIRE(rc->get(&shared_key, &got_shared) >= 1); + CHECK(std::memcmp(got_shared->data(), shared.data(), shared.size()) == 0); +}