From 328e026b321e390c1efb9e037b6d89b9ca9dff3a Mon Sep 17 00:00:00 2001 From: "Phong X. Nguyen" Date: Tue, 14 Jul 2026 15:27:43 +0000 Subject: [PATCH] Fix locking and lifetime races in CLFUS compress_entries The fastlz minimum-length check ran in the region where the stripe mutex is dropped and jumped to Lfailed, writing to the entry and advancing the compression cursor without the lock and skipping the relock, so the rest of the pass ran unlocked. The compression-failure check similarly ran before revalidating that the entry survived the unlocked compression, so a failure could mark a freed entry; it now runs after revalidation. Also guard against the compression cursor having been invalidated while the lock was dropped, which dereferenced null. Observable behavior for live entries is unchanged: undersized or uncompressible entries are marked incompressible and skipped. The races are only observable under concurrency, so the added test pins the single-threaded behavior of the seam; the class declaration moves into RamCacheCLFUS.h so the test can drive compress_entries() directly. Co-Authored-By: Claude Fable 5 --- src/iocore/cache/CMakeLists.txt | 1 + src/iocore/cache/RamCacheCLFUS.cc | 92 +++------ src/iocore/cache/RamCacheCLFUS.h | 99 ++++++++++ .../test_RamCacheCompressEntries.cc | 182 ++++++++++++++++++ 4 files changed, 304 insertions(+), 70 deletions(-) create mode 100644 src/iocore/cache/RamCacheCLFUS.h create mode 100644 src/iocore/cache/unit_tests/test_RamCacheCompressEntries.cc diff --git a/src/iocore/cache/CMakeLists.txt b/src/iocore/cache/CMakeLists.txt index f8a252b430c..5a04183c0f1 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(RamCacheCompressEntries unit_tests/test_RamCacheCompressEntries.cc) # Unit Tests without unit_tests/main.cc add_executable(test_ConfigVolumes unit_tests/test_ConfigVolumes.cc) diff --git a/src/iocore/cache/RamCacheCLFUS.cc b/src/iocore/cache/RamCacheCLFUS.cc index feb09781eb8..dcabd4c85c3 100644 --- a/src/iocore/cache/RamCacheCLFUS.cc +++ b/src/iocore/cache/RamCacheCLFUS.cc @@ -24,6 +24,7 @@ // Clocked Least Frequently Used by Size (CLFUS) replacement policy // See https://cwiki.apache.org/confluence/display/TS/RamCache +#include "RamCacheCLFUS.h" #include "P_RamCache.h" #include "P_CacheInternal.h" #include "StripeSM.h" @@ -63,27 +64,6 @@ DbgCtl dbg_ctl_ram_cache_compare{"ram_cache_compare"}; #endif -struct RamCacheCLFUSEntry { - CryptoHash key; - uint64_t auxkey; - uint64_t hits; - uint32_t size; // memory used including padding in buffer - uint32_t len; // actual data length - uint32_t compressed_len; - union { - struct { - uint32_t compressed : 3; // compression type - uint32_t incompressible : 1; - uint32_t lru : 1; - uint32_t copy : 1; // copy-in-copy-out - } flag_bits; - uint32_t flags; - }; - LINK(RamCacheCLFUSEntry, lru_link); - LINK(RamCacheCLFUSEntry, hash_link); - Ptr data; -}; - constexpr uint64_t requeue_hits(const uint64_t hits) { @@ -102,46 +82,6 @@ cache_value(const RamCacheCLFUSEntry *const e) return cache_value_hits_size(e->hits, e->size); } -class RamCacheCLFUS : public RamCache -{ -public: - RamCacheCLFUS() {} - - // returns 1 on found/stored, 0 on not found/stored, if provided auxkey1 and auxkey2 must match - int get(CryptoHash *key, Ptr *ret_data, uint64_t auxkey = 0) override; - int put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy = false, uint64_t auxkey = 0) override; - int fixup(const CryptoHash *key, uint64_t old_auxkey, uint64_t new_auxkey) override; - int64_t size() const override; - - void init(int64_t max_bytes, StripeSM *stripe) override; - - void compress_entries(EThread *thread, int do_at_most = INT_MAX); - - // TODO move it to private. - StripeSM *stripe = nullptr; // for stats -private: - int64_t _max_bytes = 0; - int64_t _bytes = 0; - int64_t _objects = 0; - - double _average_value = 0; - int64_t _history = 0; - int _ibuckets = 0; - int _nbuckets = 0; - DList(RamCacheCLFUSEntry, hash_link) *_bucket = nullptr; - Que(RamCacheCLFUSEntry, lru_link) _lru[2]; - uint16_t *_seen = nullptr; - int _ncompressed = 0; - RamCacheCLFUSEntry *_compressed = nullptr; // first uncompressed lru[0] entry - - void _resize_hashtable(); - void _victimize(RamCacheCLFUSEntry *e); - void _move_compressed(RamCacheCLFUSEntry *e); - RamCacheCLFUSEntry *_destroy(RamCacheCLFUSEntry *e); - void _requeue_victims(Que(RamCacheCLFUSEntry, lru_link) & victims); - void _tick(); // move CLOCK on history -}; - int64_t RamCacheCLFUS::size() const { @@ -475,6 +415,12 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) default: goto Lcontinue; case CACHE_COMPRESSION_FASTLZ: + if (e->len < 16) { + // fastlz cannot compress inputs this small; decide while the entry + // is still lock-protected rather than in the unlocked region below. + e->flag_bits.incompressible = 1; + goto Lcontinue; + } l = static_cast(static_cast(e->len) * 1.05 + 66); break; case CACHE_COMPRESSION_LIBZ: @@ -495,11 +441,11 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) bool failed = false; switch (ctype) { default: - goto Lfailed; + // The bound switch above filtered unknown types; this is unreachable, + // but must not jump to Lfailed from this unlocked region. + failed = true; + break; case CACHE_COMPRESSION_FASTLZ: - if (e->len < 16) { - goto Lfailed; - } if ((l = fastlz_compress(edata->data(), elen, b)) <= 0) { failed = true; } @@ -526,11 +472,10 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) #endif } MUTEX_TAKE_LOCK(stripe->mutex, thread); - // see if the entry is till around + // See if the entry is still around; it may have been freed while the + // lock was dropped, so this must be checked before anything writes to + // it (including the failure marking below). { - if (failed) { - goto Lfailed; - } uint32_t i = key.slice32(3) % this->_nbuckets; RamCacheCLFUSEntry *ee = this->_bucket[i].head; while (ee) { @@ -540,11 +485,18 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) ee = ee->hash_link.next; } if (!ee || ee != e) { - e = this->_compressed; ats_free(b); + e = this->_compressed; + if (!e) { + // The cursor was invalidated while the lock was dropped. + break; + } goto Lcontinue; } } + if (failed) { + goto Lfailed; + } if (l > required_compression * e->len) { e->flag_bits.incompressible = true; } diff --git a/src/iocore/cache/RamCacheCLFUS.h b/src/iocore/cache/RamCacheCLFUS.h new file mode 100644 index 00000000000..037e0cfb159 --- /dev/null +++ b/src/iocore/cache/RamCacheCLFUS.h @@ -0,0 +1,99 @@ +/** @file + + Clocked Least Frequently Used by Size (CLFUS) RAM cache replacement policy. + + @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. + */ + +#pragma once + +// See https://cwiki.apache.org/confluence/display/TS/RamCache + +#include "P_RamCache.h" + +#include "iocore/eventsystem/IOBuffer.h" +#include "tscore/CryptoHash.h" +#include "tscore/List.h" + +#include +#include + +class EThread; +class StripeSM; + +struct RamCacheCLFUSEntry { + CryptoHash key; + uint64_t auxkey; + uint64_t hits; + uint32_t size; // memory used including padding in buffer + uint32_t len; // actual data length + uint32_t compressed_len; + union { + struct { + uint32_t compressed : 3; // compression type + uint32_t incompressible : 1; + uint32_t lru : 1; + uint32_t copy : 1; // copy-in-copy-out + } flag_bits; + uint32_t flags; + }; + LINK(RamCacheCLFUSEntry, lru_link); + LINK(RamCacheCLFUSEntry, hash_link); + Ptr data; +}; + +class RamCacheCLFUS : public RamCache +{ +public: + RamCacheCLFUS() {} + + // returns 1 on found/stored, 0 on not found/stored, if provided auxkey1 and auxkey2 must match + int get(CryptoHash *key, Ptr *ret_data, uint64_t auxkey = 0) override; + int put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy = false, uint64_t auxkey = 0) override; + int fixup(const CryptoHash *key, uint64_t old_auxkey, uint64_t new_auxkey) override; + int64_t size() const override; + + void init(int64_t max_bytes, StripeSM *stripe) override; + + void compress_entries(EThread *thread, int do_at_most = INT_MAX); + + // TODO move it to private. + StripeSM *stripe = nullptr; // for stats +private: + int64_t _max_bytes = 0; + int64_t _bytes = 0; + int64_t _objects = 0; + + double _average_value = 0; + int64_t _history = 0; + int _ibuckets = 0; + int _nbuckets = 0; + DList(RamCacheCLFUSEntry, hash_link) *_bucket = nullptr; + Que(RamCacheCLFUSEntry, lru_link) _lru[2]; + uint16_t *_seen = nullptr; + int _ncompressed = 0; + RamCacheCLFUSEntry *_compressed = nullptr; // first uncompressed lru[0] entry + + void _resize_hashtable(); + void _victimize(RamCacheCLFUSEntry *e); + void _move_compressed(RamCacheCLFUSEntry *e); + RamCacheCLFUSEntry *_destroy(RamCacheCLFUSEntry *e); + void _requeue_victims(Que(RamCacheCLFUSEntry, lru_link) & victims); + void _tick(); // move CLOCK on history +}; diff --git a/src/iocore/cache/unit_tests/test_RamCacheCompressEntries.cc b/src/iocore/cache/unit_tests/test_RamCacheCompressEntries.cc new file mode 100644 index 00000000000..9644f5acf0b --- /dev/null +++ b/src/iocore/cache/unit_tests/test_RamCacheCompressEntries.cc @@ -0,0 +1,182 @@ +/** @file + + Catch-based regression tests for RamCacheCLFUS::compress_entries(). + + @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. + */ + +#include "main.h" + +#include "../RamCacheCLFUS.h" +#include "../P_CacheInternal.h" + +#include +#include +#include +#include + +// Required by main.h +int cache_vols = 1; +bool reuse_existing_cache = false; + +namespace +{ + +// 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 CLFUS get/put/compress paths touch only these metrics and the stripe +// mutex, so that is all the stripe needs for these tests. +void +wire_stripe(StripeSM &stripe, CacheVol &cache_vol) +{ + stripe.cache_vol = &cache_vol; + + cache_rsb.ram_cache_bytes = ts::Metrics::Gauge::createPtr("unit_test.clfus.ram_cache.bytes"); + cache_rsb.ram_cache_hits = ts::Metrics::Counter::createPtr("unit_test.clfus.ram_cache.hits"); + cache_rsb.ram_cache_misses = ts::Metrics::Counter::createPtr("unit_test.clfus.ram_cache.misses"); + cache_vol.vol_rsb.ram_cache_bytes = ts::Metrics::Gauge::createPtr("unit_test.clfus.vol.ram_cache.bytes"); + cache_vol.vol_rsb.ram_cache_hits = ts::Metrics::Counter::createPtr("unit_test.clfus.vol.ram_cache.hits"); + cache_vol.vol_rsb.ram_cache_misses = ts::Metrics::Counter::createPtr("unit_test.clfus.vol.ram_cache.misses"); +} + +RamCacheCLFUS * +make_cache(StripeSM &stripe) +{ + // Initialize with compression disabled so init() does not schedule the + // background compressor continuation, which would retain a pointer to the + // cache; compression is driven synchronously by the tests instead. The + // caches are kept reachable for the life of the process because the policy + // has no destructor (entries are pool-allocated). + cache_config_ram_cache_compress = CACHE_COMPRESSION_NONE; + cache_config_ram_cache_compress_percent = 100; + cache_config_ram_cache_use_seen_filter = 0; + + static std::vector &all_caches = *new std::vector; + auto *rc = new RamCacheCLFUS; + + all_caches.push_back(rc); + rc->init(1 << 20, &stripe); + return rc; +} + +std::vector +pattern_bytes(std::size_t len) +{ + std::vector bytes(len); + for (std::size_t i = 0; i < len; i++) { + bytes[i] = static_cast('A' + (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; +} + +} // namespace + +// fastlz cannot compress inputs under 16 bytes. Such entries must be decided +// under the stripe lock and left intact. The races fixed alongside this test +// are only observable under concurrency, so this pins the single-threaded +// behavior of the restructured code: undersized entries are marked +// incompressible while locked, skipped on later passes, and stay readable. +TEST_CASE("CLFUS compress_entries leaves undersized payloads intact", "[cache][ramcache][compress]") +{ + CacheDisk disk; + init_disk(disk); + StripeSM stripe{&disk, 10, 0}; + CacheVol cache_vol; + wire_stripe(stripe, cache_vol); + + auto *rc = make_cache(stripe); + auto payload = pattern_bytes(8); + auto buf = make_buffer(payload); + auto key = fresh_key(); + + REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1); + + cache_config_ram_cache_compress = CACHE_COMPRESSION_FASTLZ; + // Two passes: the first marks the entry incompressible; the second must + // skip it (on the former code, the first pass's loop tail ran unlocked). + rc->compress_entries(this_ethread(), INT_MAX); + rc->compress_entries(this_ethread(), INT_MAX); + + Ptr got; + + REQUIRE(rc->get(&key, &got) == RAM_HIT_COMPRESS_NONE); + REQUIRE(got.get() != nullptr); + CHECK(std::memcmp(got->data(), payload.data(), payload.size()) == 0); +} + +TEST_CASE("CLFUS compress_entries still compresses eligible payloads", "[cache][ramcache][compress]") +{ + CacheDisk disk; + init_disk(disk); + StripeSM stripe{&disk, 10, 0}; + CacheVol cache_vol; + wire_stripe(stripe, cache_vol); + + auto *rc = make_cache(stripe); + auto payload = pattern_bytes(8192); + auto buf = make_buffer(payload); + auto key = fresh_key(); + + REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1); + + cache_config_ram_cache_compress = CACHE_COMPRESSION_FASTLZ; + int64_t size_before = rc->size(); + + rc->compress_entries(this_ethread(), INT_MAX); + CHECK(rc->size() < size_before); + + Ptr got; + + REQUIRE(rc->get(&key, &got) == RAM_HIT_COMPRESS_FASTLZ); + REQUIRE(got.get() != nullptr); + CHECK(std::memcmp(got->data(), payload.data(), payload.size()) == 0); +}