diff --git a/src/iocore/cache/CMakeLists.txt b/src/iocore/cache/CMakeLists.txt index f8a252b430c..daa0f10b297 100644 --- a/src/iocore/cache/CMakeLists.txt +++ b/src/iocore/cache/CMakeLists.txt @@ -92,6 +92,7 @@ if(BUILD_TESTING) add_cache_test(Update_Header unit_tests/test_Update_header.cc) add_cache_test(CacheStripe unit_tests/test_Stripe.cc) add_cache_test(CacheAggregateWriteBuffer unit_tests/test_AggregateWriteBuffer.cc) + add_cache_test(RamCacheCopy unit_tests/test_RamCacheCopy.cc) # Unit Tests without unit_tests/main.cc add_executable(test_ConfigVolumes unit_tests/test_ConfigVolumes.cc) diff --git a/src/iocore/cache/P_RamCache.h b/src/iocore/cache/P_RamCache.h index 1afe1037517..b1bb64a0184 100644 --- a/src/iocore/cache/P_RamCache.h +++ b/src/iocore/cache/P_RamCache.h @@ -27,6 +27,9 @@ #include "iocore/eventsystem/IOBuffer.h" #include "tscore/CryptoHash.h" +#include "tscore/ink_memory.h" + +#include class StripeSM; @@ -41,6 +44,36 @@ 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) + { + char *b = static_cast(ats_malloc(len)); + + memcpy(b, data->data(), len); + Ptr d = make_ptr(new_xmalloc_IOBufferData(b, len)); + + d->_mem_type = DEFAULT_ALLOC; + 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 639d2faa33c..09102605468 100644 --- a/src/iocore/cache/RamCacheCLFUS.cc +++ b/src/iocore/cache/RamCacheCLFUS.cc @@ -314,12 +314,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); @@ -624,11 +623,8 @@ 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; } check_accounting(this); e->flag_bits.copy = copy; @@ -735,10 +731,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 6e392338be5..0336da0dcb6 100644 --- a/src/iocore/cache/RamCacheLRU.cc +++ b/src/iocore/cache/RamCacheLRU.cc @@ -33,6 +33,7 @@ struct RamCacheLRUEntry { CryptoHash key; uint64_t auxkey; + bool copy; // copy-in-copy-out: buffers are never shared with callers LINK(RamCacheLRUEntry, lru_link); LINK(RamCacheLRUEntry, hash_link); Ptr data; @@ -147,7 +148,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->data->block_size()); + } 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 +186,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; @@ -213,6 +217,18 @@ RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t if (e->auxkey == auxkey) { lru.remove(e); lru.enqueue(e); + if (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. + int64_t delta = static_cast(len) - e->data->block_size(); + + e->data = copy_data_in(data, 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); @@ -224,13 +240,18 @@ 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->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..f5a70ab1017 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,23 @@ RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32 if (e->freq < FREQ_MAX) { e->freq++; } + if (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. + 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 +420,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..4530aa671b7 --- /dev/null +++ b/src/iocore/cache/unit_tests/test_RamCacheCopy.cc @@ -0,0 +1,260 @@ +/** @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 "../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 +{ + +struct PolicyCase { + RamCache *(*factory)(); + const char *name; +}; + +const PolicyCase policy_cases[] = { + {new_RamCacheLRU, "LRU" }, + {new_RamCacheCLFUS, "CLFUS" }, + {new_RamCacheS3FIFO, "S3FIFO"}, +}; + +// Minimal CacheDisk wiring needed to construct a StripeSM. Mirrors the helper +// in test_Stripe.cc. +void +init_disk(CacheDisk &disk) +{ + disk.path = static_cast(ats_malloc(1)); + disk.path[0] = '\0'; + disk.disk_stripes = static_cast(ats_malloc(sizeof(DiskStripe *))); + disk.disk_stripes[0] = nullptr; + disk.header = static_cast(ats_malloc(sizeof(DiskHeader))); + disk.header->num_volumes = 0; +} + +// 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; +} + +CryptoHash +fresh_key() +{ + static uint64_t salt = 0; + + ++salt; + CryptoHash key; + key.u64[0] = 0xc0ffee00 + salt; + key.u64[1] = 0xdeadbeef + salt; + return key; +} + +RamCache * +make_cache(const PolicyCase &pc, StripeSM &stripe) +{ + // 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 = pc.factory(); + + all_caches.push_back(rc); + rc->init(1 << 20, &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, stripe); + auto payload = pattern_bytes(8192, '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, stripe); + auto payload = pattern_bytes(8192, '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, stripe); + auto payload = pattern_bytes(8192, '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, stripe); + auto payload = pattern_bytes(8192, '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()); +}